Striim 3.9.4 / 3.9.5 documentation

DATA()

For each event, returns the values in the WAEvent data array stream as a java.util.HashMap, with column names as the keys.

The following example shows how to use the DATA() function to include database column names from a CDC source in JSONFormatter output. (You could do the same thing with AVROFormatter.)

CREATE SOURCE DBROracleIn USING DatabaseReader (
  Username:'striidbr',
  Password:'passwd',
  ConnectionURL:'jdbc:oracle:thin:@192.0.2.49:1521/orcl',
  Tables:'ROBERT.POSAUTHORIZATIONS',
  FetchSize:1
) 
OUTPUT TO OracleRawStream;

CREATE TYPE OpTableDataType(
  TableName String,
  data java.util.HashMap
);
CREATE STREAM OracleTypedStream OF OpTableDataType;
CREATE CQ ParseOracleRawStream
  INSERT INTO OracleTypedStream
  SELECT META(OracleRawStream, "TableName").toString(),
    DATA(OracleRawStream)
  FROM OracleRawStream;
 
CREATE TARGET DBR2JSONOut USING FileWriter(
  filename:'DBR2JSON.json'
)
FORMAT USING JSONFormatter ()
INPUT FROM OracleTypedStream;

The CQ will be easier to read if you use an alias for the stream name. For example:

CREATE CQ ParseOracleRawStream
  INSERT INTO OracleTypedStream
  SELECT META(x, "TableName").toString(),
    DATA(x)
  FROM OracleRawStream x;

Assuming the following Oracle table and data:

CREATE TABLE POSAUTHORIZATIONS (
  BUSINESS_NAME varchar2(30),
  MERCHANT_ID varchar2(100),
  PRIMARY_ACCOUNT NUMBER,
  POS NUMBER,CODE varchar2(20),
  EXP char(4),
  CURRENCY_CODE char(3),
  AUTH_AMOUNT number(10,3),
  TERMINAL_ID NUMBER,
  ZIP number,
  CITY varchar2(20));
commit;
INSERT INTO POSAUTHORIZATIONS VALUES(
  'COMPANY 1',
  'D6RJPwyuLXoLqQRQcOcouJ26KGxJSf6hgbu',
  6705362103919221351,
  0,
  '20130309113025',
  '0916',
  'USD',
  2.20,
  5150279519809946,
  41363,
  'Quicksand');
commit;

Output for this application would be:

{
 "TableName":"MYSCHEMA.POSAUTHORIZATIONS",
 "data":{"AUTH_AMOUNT":"2.2", "BUSINESS_NAME":"COMPANY 1", "ZIP":"41363", "EXP":"0916", 
"POS":"0", "CITY":"Quicksand", "CURRENCY_CODE":"USD", 
"PRIMARY_ACCOUNT":"6705362103919221351", 
"MERCHANT_ID":"D6RJPwyuLXoLqQRQcOcouJ26KGxJSf6hgbu", "TERMINAL_ID":"5150279519809946",
"CODE":"20130309113025"}
}