About change data capture (CDC)

Relational database management systems use write-ahead logs, also called redo or transaction logs, that represent DML and DDL changes.  Traditionally, RDBMS systems use these logs to guarantee ACID properties and support rollback and roll-forward recovery operations.  As DBMS technology has evolved, these logs have been augmented to record additional types of changes. Today they may track virtually every redoable and undoable action in the system, including transaction start and commit boundaries, table and index changes, data definition changes, rollback operations, indicators of non-logged changes, and more.

DBMS vendors and third parties have found additional uses for these logs. Striim, for example, can extract change data from logs in real time in order to make information available before the DBMS has finished processing it, at the same time minimizing the performance load on the RBMS by eliminating additional queries. There are many potential uses for this information, such raising alerts about error conditions sooner and double-checking DBMS operations in order to identify lost data.

All of the readers discussed in this Change Data Capture Guide capture change data by reading logs.

You can use Striim's Incremental Batch Reader to capture change data using JDBC based on timestamps or incrementing values.IncrementalBatchReader

The output data type for sources that use change data capture readers is WAEvent. The fields and valid values vary among the readers, but they all include the following:

metadata: a map including the elements:

To retrieve the values for these fields, use the META() function. See Parsing the fields of WAEvent for CDC readers.

data: an array of fields, numbered from 0, containing:

To retrieve the values for these fields, use SELECT ... (DATA[]). See Parsing the fields of WAEvent for CDC readers.

before (for UPDATE operations only): the same format as data, but containing the values as they were prior to the UPDATE operation

dataPresenceBitMap, beforePresenceBitMap , and typeUUID are reserved and should be ignored.

For information on additional fields and detailed discussion of values, see:

The following sample application uses OracleReader but the approach is the same for all CDC readers.

CREATE APPLICATION SampleCDCApp;
CREATE SOURCE OracleCDCIn USING OracleReader (
  Username:'striim',
  Password:'passwd',
  ConnectionURL:'203.0.113.49:1521:orcl',
  Tables:'MYSCHEMA.POSAUTHORIZATIONS',
  FetchSize:1
) 
OUTPUT TO OracleCDCStream;
 
CREATE TYPE PosMeta(
 tableName String,
 operationName String,
 txnID String,
 timestamp String
);
CREATE STREAM PosMetaStream OF PosMeta; 
CREATE TYPE PosData(
  businessName String,
  accountName String,
  pos String,
  code String
);
CREATE STREAM PosDataStream OF PosData;
-- extract the metadata values
CREATE CQ OracleToPosMeta
INSERT INTO PosMetaStream
SELECT
    META(m,"TableName").toString(),
    META(m,"OperationName").toString(),
    META(m,"TxnID").toString(),
    META(m,"TimeStamp").toString()
       FROM OracleCDCStream m;
-- write the metadata values to SysOut
CREATE TARGET Metadump USING SysOut(name:meta) INPUT FROM PosMetaStream;
 
-- extract the data values
CREATE CQ OracleToPosData
INSERT INTO PosDataStream
SELECT
    CASE WHEN IS_PRESENT(x,data,0)==true THEN data[0].toString()
        ELSE "NOT_PRESENT"
    END,
    CASE WHEN IS_PRESENT(x,data,1)==true THEN data[1].toString()
        ELSE "NOT_PRESENT"
    END,
    CASE WHEN IS_PRESENT(x,data,2)==true THEN data[2].toString()
        ELSE "NOT_PRESENT"
    END,
    CASE WHEN IS_PRESENT(x,data,3)==true THEN data[3].toString()
        ELSE "NOT_PRESENT"
    END
FROM OracleCDCStream x;
-- write dump the data values to SysOut
CREATE TARGET  Datadump USING SysOut(name:data) INPUT FROM PosDataStream;
END APPLICATION SampleCDCApp;

The output for the three operations described in OracleReader example output would be similar to:

meta: PosMeta_1_0{
  tableName: "SCOTT.POSAUTHORIZATIONS"
  operationName: "INSERT"
  txnID: "4.0.1742"
  timestamp: "2015-12-11T16:31:30.000-08:00"
};
data: PosData_1_0{
  businessName: "COMPANY 1"
  accountName: "D6RJPwyuLXoLqQRQcOcouJ26KGxJSf6hgbu"
  pos: "6705362103919221351"
  code: "0"
};
meta: PosMeta_1_0{
  tableName: "SCOTT.POSAUTHORIZATIONS"
  operationName: "UPDATE"
  txnID: "4.0.1742"
  timestamp: "2015-12-11T16:31:30.000-08:00"
};
data: PosData_1_0{
  businessName: "COMPANY 5A"
  accountName: "D6RJPwyuLXoLqQRQcOcouJ26KGxJSf6hgbu"
  pos: "6705362103919221351"
  code: "0"
};
meta: PosMeta_1_0{
  tableName: "SCOTT.POSAUTHORIZATIONS"
  operationName: "DELETE"
  txnID: "4.0.1742"
  timestamp: "2015-12-11T16:31:30.000-08:00"
};
data: PosData_1_0{
  businessName: "COMPANY 5A"
  accountName: "D6RJPwyuLXoLqQRQcOcouJ26KGxJSf6hgbu"
  pos: "6705362103919221351"
  code: "0"
};

In this release, table mapping is validated only for applications with a single MariaDB, MySQL, Oracle, PostgreSQL, or SQL Server source (DatabaseReader, IncrementalBatchReader, MySQLReader, OracleReader, PostgreSQLReader, or MSSQLReader) and a single MariaDB, MySQL, Oracle, PostgreSQL, or SQL Server DatabaseWriter target.

When an application is deployed, Striim will compare the source and target columns and pop up a Validation errors dialog if it finds any of the following:

For example:

When you see this dialog, you may:

After you have made corrections, choose Validation Errors from the Created menu and click Validate Again.

HP NonStop readers, MSSQLReader, and OracleReader can all read data from multiple tables using a single source. The MAP function allows data from each table to be output to a different stream.

The following reads all tables from the schema SCOTT.TEST.

CREATE SOURCE OraSource USING OracleReader (
  ... Tables:'SCOTT.%' ...

The following reads from all tables with names that start with SCOTT.TEST, such as SCOTT.TEST1, SCOTT.TESTCASE, and so on.

CREATE SOURCE OraSource USING OracleReader (
  ... Tables:'SCOTT.TEST%' ...

The following reads all tables with names that start with S and end with .TEST. Again, any tables added after the application starts will be ignored.

CREATE SOURCE OraSource USING OracleReader (
  ... Tables:='S%.TEST' ...

The following shows how to query data from one table when a stream contains data from multiple tables.

CREATE SOURCE OraSource USING OracleReader (
  ... Tables:'SCOTT.POSDATA;SCOTT.STUDENT' ...
)
OUTPUT TO Orders;
...
CREATE CQ renderOracleControlLogEvent
INSERT INTO oracleControlLogStream
    META(x,"OperationName"),
    META(x,"TxnID"),
    META(x,"TimeStamp").toString(),
    META(x,"TxnUserID”),
   data[0] 
FROM Orders x
WHERE META(x,”TableName").toString() = "SCOTT.POSDATA";

The following takes input from two tables and sends output for each to a separate stream using the MAP function. Note that a regular output stream (in this case Orders) must also be specified, even if it is not used by the application.

CREATE SOURCE OraSource USING OracleReader (
  ... Tables:'SCOTT.POSDATA;SCOTT.STUDENT' ...
)
OUTPUT TO OrderStream,
  PosDataStream MAP (table:'SCOTT.POSDATA'),
  StudentStream MAP (table:'SCOTT.STUDENT');

In some cases, creating a separate source for each table may improve performance.

When a SQL CDC source reads from multiple tables, you may use OUTPUT TO <stream name> MAP (Table:'<table name>') to route events from each table to a different output stream. Striim will create a type for each stream using the column names and data types of the source table. Date and time data types are mapped to DateTime. All other types are mapped to String.

In this release, OUTPUT TO ... MAP is not supported for PostgreSQLReader.

The following takes input from two tables and sends output for each to a separate stream using the MAP function. Note that a regular, unmapped output stream (in this case OrderStream) must also be specified, even if all tables are mapped.

CREATE SOURCE OraSource USING OracleReader (
  ... Tables:'SCOTT.POSDATA;SCOTT.STUDENT' ...
)
OUTPUT TO OrderStream,
OUTPUT TO PosDataStream MAP (Table:'SCOTT.POSDATA'),
OUTPUT TO StudentStream MAP (Table:'SCOTT.STUDENT');

In some cases, creating a separate source for each table may improve performance over using OUTPUT TO ... MAP.

Use the PutUserData function in a CQ to add an element to the WAEvent USERDATA map. Elements in USERDATA may, for example, be inserted into DatabaseWriter output as described in Modifying output using ColumnMap or used to partition KafkaWriter output among multiple partitions (see discussion of the Partition Key property in Kafka Writer).

The following example would add the sixth element (counting from zero) in the WAEvent data array to USERDATA as the field "city":

CREATE CQ AddUserData
INSERT INTO OracleSourceWithPartitionKey
SELECT putUserData(x, 'city',data[5])
FROM OracleSourcre_ChangeDataStream x;

You may add multiple fields, separated by commas:

SELECT putUserData(x, 'city',data[5], 'state',data[6])

For examples of how to use Userdata elements in TQL, see Modifying output using ColumnMap and the discussions of PartitionKey in Kafka Writer and S3 Writer.

To remove an element from USERDATA, use the removeUserData function (you may specify multiple elements, separated by commas):

CREATE CQ RemoveUserData
INSERT INTO OracleSourceWithPartitionKey
SELECT removeUserData(x, 'city')
FROM OracleSourcre_ChangeDataStream x;

To remove all elements from the USERDATA map, use the clearUserData function:

CREATE CQ ClearUserData
INSERT INTO OracleSourceWithPartitionKey
SELECT clearUserData(x)
FROM OracleSourcre_ChangeDataStream x;

When a CDC reader's output is the input of a writer, you may insert a CQ between the two to modify the WAEvent's data array using the following functions. This provides more flexibility when replicating data.

replaceData()

replaceData(WAEvent s, String 'columnName', Object o)

For input stream s, replaces the data array value for a specified column with an object. The object must be of the same type as the column.

For example, the following would replace the value of the DESCRIPTION column with the string redacted:

CREATE CQ replaceDataCQ
INSERT INTO opStream
SELECT replaceData(s,'DESCRIPTION','redacted')
FROM OracleReaderOutput s;

Optionally, you may restrict the replacement to a specific column:

replaceData(WAEvent s, String 'tableName', String 'columnName', Object o)

replaceString()

replaceString(WAEvent s, String 'findString', String 'newString') 

For input stream s, replaces all occurrences of findString in the data array with newString. For example, the following would replace all occurrences of MyCompany with PartnerCompany:

CREATE CQ replaceDataCQ
INSERT INTO opStream
SELECT replaceString(s,'MyCompany','PartnerCompany')
FROM OracleReaderOutput s;

replaceStringRegex()

replaceStringRegex(WAEvent s, String 'regex', String 'newString')

For input stream s, replaces all strings in the data array that match the regex expression with newString. For example, the following would remove all whitespace:

CREATE CQ replaceDataCQ
INSERT INTO opStream
SELECT replaceStringRegex(s,’\\\\s’,’’)
FROM OracleReaderOutput s;

The following would replace all numerals with x:

CREATE CQ replaceDataCQ
INSERT INTO opStream
SELECT replaceStringRegex(s,’\\\\d’,’x’)
FROM OracleReaderOutput s;

When a CDC reader's output is the input of a writer, you may insert a CQ between the two to modify the values in the WAEvent's data array. This provides more flexibility when replicating data.

In this context, the syntax for the SELECT statement is:

SELECT <alias> FROM <stream name> <alias> MODIFY (<alias>.data[<field number>] = <expression>,...)

The following simple example would convert a monetary amount in the data[4] field using an exchange rate of 1.09:

CREATE STREAM ConvertedStream OF TYPE Global.WAEvent;
CREATE CQ ConvertAmount 
INSERT INTO ConvertedStream
SELECT r FROM RawStream r
MODIFY(r.data[4] = TO_FLOAT(r.data[4]) * 1.09);

The next example illustrates the use of masking functions and CASE statements. It uses the maskPhoneNumber function (see Masking functions) to mask individually identifiable information from US and India telephone numbers (as dialed from the US) while preserving the country and area codes. The US numbers have the format ###-###-####, where the first three digits are the area code. India numbers have the format 91-###-###-####, where 91 is the country code and the third through fifth digits are the subscriber trunk dialing (STD) code. The telephone numbers are in data[4] and the country codes are in data[5].

CREATE STREAM MaskedStream OF Global.WAEvent;
CREATE CQ maskData 
INSERT INTO maskedDataStream
SELECT r FROM RawStream r
MODIFY(
r.data[4] = CASE
    WHEN TO_STRING(r.data[5]) == "US" THEN maskPhoneNumber(TO_STRING(r.data[4]), "###-xxx-xxx")
    ELSE maskPhoneNumber(TO_STRING(r.data[4]), "#####x#xxx#xxxx")
  END
);

This could be extended with additional WHEN statements to mask numbers from additional countries, or with additional masking functions to mask individually identifiable information such as credit card, Social Security, and national identification numbers.

See Masking functions for additional examples.

The DATA() and BEFORE() functions return the WAEvent data and before arrays. The following example shows how you could use these functions to write change data event details to a JSON file with the associated Oracle column names. (You could do the same thing with AVROFormatter.) This is supported only for 11g using LogMiner.

DATAORDERED(x) and BEFOREORDERED() return the column values in the same order as in the source table. When using DATA(x) and BEFORE(x), the order is not guaranteed.

CREATE SOURCE OracleCDCIn USING OracleReader (
  Username:'walm',
  Password:'passwd',
  ConnectionURL:'192.168.1.49:1521:orcl',
  Tables:'myschema.%',
  FetchSize:1
) 
OUTPUT TO OracleRawStream;CREATE TYPE OpTableDataType(
  OperationName String,
  TableName String,
  data java.util.HashMap,
  before java.util.HashMap
);

CREATE STREAM OracleTypedStream OF OpTableDataType;
CREATE CQ ParseOracleRawStream
  INSERT INTO OracleTypedStream
  SELECT META(OracleRawStream, "OperationName").toString(),
    META(OracleRawStream, "TableName").toString(),
    DATA(OracleRawStream),
    BEFORE(OracleRawStream)
  FROM OracleRawStream;
 
CREATE TARGET OracleCDCFFileOut USING FileWriter(
  filename:'Oracle2JSON_withFFW.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, "OperationName").toString(),
    META(x, "TableName").toString(),
    DATA(x),
    BEFORE(x)
  FROM OracleRawStream x;

Using this application, the output for the INSERT operation described in OracleReader example output would look like this:

 {
  "OperationName":"UPDATE",
  "TableName":"ROBERT.POSAUTHORIZATIONS",
  "data":{"AUTH_AMOUNT":"2.2", "BUSINESS_NAME":"COMPANY 5A", "ZIP":"41363", "EXP":"0916", 
"POS":"0", "CITY":"Quicksand", "CURRENCY_CODE":"USD", "PRIMARY_ACCOUNT":"6705362103919221351", 
"MERCHANT_ID":"D6RJPwyuLXoLqQRQcOcouJ26KGxJSf6hgbu", "TERMINAL_ID":"5150279519809946", 
"CODE":"20130309113025"},
  "before":{"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"}
 }

When replicating CDC source data with Azure Synapse Writer, BigQuery Writer, Cosmos DB Writer, Database Writer, Databricks Writer, Fabric Data Warehouse Writer, MongoDB Cosmos DB Writer, MongoDB Writer, or Spanner Writer, attempted updates to and deletes from the target database may sometimes fail due to missing tables, duplicate or missing primary keys, or other issues. See CREATE EXCEPTIONSTORE for discussion of how to ignore these errors and capture the unwritten events.

See Handling schema evolution.

The first step in Oracle-to-Oracle replication is the initial load.

Once initial load is complete, the following sample application would continuously replicate changes to the tables SOURCE1 and SOURCE2 in database DB1 to tables TARGET1 and TARGET2 in database DB2 using Database Writer. The StartSCN value is the SCN number from step 1. In  the WHERE clause, replace  ######### with the SCN from step 2. Start the application with recovery enabled (see Recovering applications) so that on restart it will resume from the latest transaction rather than the StartSCN point.Recovering applications

CREATE SOURCE OracleCDC USING OracleReader (
  Username:'striim',
  Password:'******', 
  ConnectionURL:'10.211.55.3:1521:orcl1',
  Tables:'DB1.SOURCE1;DB1.SOURCE2', 
  Compression:true
  StartSCN:'...'
)
OUTPUT TO OracleCDCStream;

CREATE CQ FilterCDC
INSERT INTO FilteredCDCStream
SELECT x 
FROM OracleCDCStream x
WHERE TO_LONG(META(x,'COMMITSCN')) > #########;

CREATE TARGET WriteToOracle USING DatabaseWriter ( 
  ConnectionURL:'jdbc:oracle:thin:@10.211.55.3:1521:orcl1', 
  Username:'striim',
  Password:'******', 
  Tables:'DB1.SOURCE1,DB2.TARGET1;DB1.SOURCE2,DB2.TARGET2'
)
INPUT FROM FilteredCDCStream;

The FilterCDC CQ filters out all transactions that were replicated during initial load.

The following Oracle column types are supported:

Limitations:

Striim provides a wizard for creating applications that read from Oracle and write to Redshift. See Creating an application using a wizard for details.

RedshiftWriter can continuously replicate one or many Oracle tables to an Amazon Redshift store. First, create a table in Redshift corresponding to each Oracle table to be replicated. Then load the existing data using DatabaseReader, for example:

CREATE SOURCE OracleJDBCSource USING DatabaseReader (
  Username:'Striim',
  Password:'****',
  ConnectionURL:'jdbc:oracle:thin:@192.168.123.14:1521/XE',
  Tables:'TPCH.H_CUSTOMER;TPCH.H_PART;TPCH.H_SUPPLIER'
)
OUTPUT TO DataStream;

CREATE TARGET TPCHInitialLoad USING RedshiftWriter (
  ConnectionURL: 'jdbc:redshift://mys3bucket.c1ffd5l3urjx.us-west-2.redshift.amazonaws.com:5439/dev',
  Username:'mys3user',
  Password:'******',
  bucketname:'mys3bucket',
/* for striimuser */
  accesskeyid:'********************',
  secretaccesskey:'****************************************',
  Tables:'TPCH.H_CUSTOMER,customer;TPCH.H_PART,part;TPCH.H_SUPPLIER,supplier'
)
INPUT FROM DataStream; 

Theœ Tables property maps each specified Oracle table to a Redshift table, for example, TPCH.H_CUSTOMER to customer.

Once the initial load is complete, the following application will read new data using LogMiner and continuously replicate it to Redshift:

CREATE SOURCE OracleCDCSource USING OracleReader (
  Username:'miner',
  Password:'miner',
  ConnectionURL:'192.168.123.26:1521:XE',
  Tables:'TPCH.H_CUSTOMER;TPCH.H_PART;TPCH.H_SUPPLIER'
)
Output To LCRStream;

CREATE TARGET RedshiftTarget USING RedshiftWriter (
  ConnectionURL: 'jdbc:redshift://mys3bucket.c1ffd5l3urjx.us-west-2.redshift.amazonaws.com:5439/dev',
  Username:'mys3user',
  Password:'******',
  bucketname:'mys3bucket',
/* for striimuser */
  accesskeyid:'********************',
  secretaccesskey:'****************************************',
  Tables:'TPCH.H_CUSTOMER,customer;TPCH.H_PART,part;TPCH.H_SUPPLIER,supplier',
  Mode:'IncrementalLoad' )
INPUT FROM LCRStream; 

Note that Redshift does not enforce unique primary key constraints. Use OracleReader's StartSCN or StartTimestamp property to ensure that you do not have duplicate or missing events in Redshift.

For for more information, see Redshift Writer.

CosmosDBWriter can continuously replicate one or many Oracle tables to Cosmos DB collections.

You must create the target collections in Cosmos DB manually. Each partition key name must match one of the column names in the Oracle source table.

Striim provides a wizard for creating applications that read from Oracle and write to Cosmos DB. See Creating an application using a wizard for details.

If you wish to run the following examples, adjust the Oracle Reader properties and Cosmos DB Writer properties to reflect your own environment.OracleReader properties

Cassandra Writer can continuously replicate one or many Oracle tables to a Cassandra or DataStax keyspace. First, create a table in Cassandra corresponding to each Oracle table to be replicated. Then load the existing data using DatabaseReader, for example:

CREATE SOURCE OracleJDBCSource USING DatabaseReader (
  Username:'Striim',
  Password:'****',
  ConnectionURL:'jdbc:oracle:thin:@192.168.123.14:1521/XE',
  Tables:'TPCH.H_CUSTOMER;TPCH.H_PART;TPCH.H_SUPPLIER'
)
OUTPUT TO DataStream;

CREATE TARGET TPCHInitialLoad USING CassandraWriter (
  ConnectionURL:'jdbc:cassandra://203.0.113.50:9042/mykeyspace',
  Username:'striim',
  Password:'******',
  Tables:'TPCH.H_CUSTOMER,customer;TPCH.H_PART,part;TPCH.H_SUPPLIER,supplier'
)
INPUT FROM DataStream; 

DatabaseWriter's Tables property maps each specified Oracle table to a Cassandra table, for example, TPCH.H_CUSTOMER to customer. Oracle table names must be uppercase and Cassandra table names must be lowercase. Since columns in Cassandra tables are not created in the same order they are specified in the CREATE TABLE statement, the ColumnMap option is required (see Mapping columns) and wildcards are not supported. See Database Reader and Cassandra Writer for more information about the properties.

Once the initial load is complete, the following application will read new data using LogMiner and continuously replicate it to Cassandra:

CREATE SOURCE OracleCDCSource USING OracleReader ( 
  Username: 'Striim',
  Password: '******',
  ConnectionURL: '203.0.113.49:1521:orcl',
  Compression:'True',
  Tables: 'TPCH.H_CUSTOMER;TPCH.H_PART;TPCH.H_SUPPLIER'
 ) 
OUTPUT TO DataStream;

CREATE TARGET CassandraTarget USING CassandraWriter(
  ConnectionURL:'jdbc:cassandra://203.0.113.50:9042/mykeyspace',
  Username:'striim',
  Password:'******',
  Tables: 'TPCH.H_CUSTOMER,customer;TPCH.H_PART,part;TPCH.H_SUPPLIER,supplier'
INPUT FROM DataStream;

OracleReader's Compression property must be True. Cassandra does not allow primary key updates. See Oracle Reader properties and Cassandra Writer for more information about the properties.

When the input stream of a Cassandra Writer target is the output of an Oracle source (DatabaseReader or OracleReader), the following types are supported:

Striim provides a wizard for creating applications that read from Oracle and write to BigQuery. See Creating an application using a wizard for details.

The following application will replicate data from all tables in MYSCHEMA in Oracle to the corresponding tables in mydataset in BIgQuery. The tables in BigQuery must exist when the application is started. All source and target tables must all have a UUID column. In the source tables, the UUID values must be unique identifiers. See the notes for the Mode property in BigQuery Writer) for additional details.Big Query Writer

CREATE SOURCE OracleCDCSource USING OracleReader ( 
  CommittedTransactions: false,
  Username: 'myuser',
  Password: 'mypass',
  ConnectionURL: '192.168.33.10:1521/XE',
  Tables: 'MYSCHEMA.%'
 ) 
OUTPUT TO DataStream;

CREATE TARGET BigQueryTarget USING BigQueryWriter(
  ServiceAccountKey: '<path>/<configuration file>.json',
  ProjectId: '<project ID>',
  Mode: 'MERGE',
  Tables: "MYSCHEMA.%,mydataset.% keycolumns(UUID)"
INPUT FROM DataStream;

See BigQuery Writer for details of the property values.

See Migrating an Oracle database to Cloud SQL for PostgreSQL using Striim.

See Continuous data replication to Cloud Spanner using Striim.

Striim provides a wizard for creating applications that read from Oracle and write to Hazelcast. See Creating an application using a wizard for details.

See Hazelcast Writer for information on the adapter properties.

To replicate Oracle data to Hazelcast:

This example assumes the following Oracle table definition: 

CREATE TABLE INV ( 
  SKU INT PRIMARY KEY NOT NULL,
  STOCK NUMBER(*,4),
  NAME varchar2(20),
  LAST_UPDATED date 
);

The following Java class defines a POJO corresponding to the table:

package com.customer.vo;
import java.io.Serializable;
import java.util.Date;
public class ProductInvObject  implements Serializable {

   public long sku = 0;
   public double stock = 0;
   public String name = null;
   public Date lastUpdated = null;

   public ProductInvObject ( ) {    }

   @Override
   public String toString() {
       return "sku : " + sku + ", STOCK:" + stock + ", NAME:" + name + ", LAST_UPDATED:" + lastUpdated  ;
   }
}

The following ORM file maps the Oracle table columns to Hazelcast maps:

<?xml version="1.0" encoding="UTF-8"?>
<entity-mappings xmlns="http://www.eclipse.org/eclipselink/xsds/persistence/orm" version="2.4">

    <entity name="prodcInv" class="com.customer.vo.ProductInvObject" >  
        <table name="MYSCHEMA.INV"/>
        <attributes>
            <id name ="sku" attribute-type="long"  >
                <column nullable="false" name="SKU" />
            </id>
            <basic name="stock" attribute-type="double" >
                <column nullable="false" name="STOCK"  />
            </basic>
            <basic name="name" attribute-type="String" >
                <column  name="NAME" />
            </basic>
            <basic name="lastUpdated"  attribute-type="java.util.Date" >
                <column name="LAST_UPDATED" />
            </basic>
        </attributes>
    </entity>

</entity-mappings>

 Assuming that the ORM file has been saved to Striim/Samples/Ora2HCast/invObject_orm.xml, the following Striim application will perform the initial load:

CREATE APPLICATION InitialLoadOra2HC;

CREATE SOURCE OracleJDBCSource USING DatabaseReader (
  Username:'striim',
  Password:'passwd',
  ConnectionURL:'203.0.113.49:1521:orcl',
  Tables:'MYSCHEMA.INV'
)
OUTPUT TO DataStream;

CREATE TARGET HazelOut USING HazelcastWriter (
  ConnectionURL: '203.0.1113.50:5702',
  ormFile:"Samples/Ora2HCast/invObject_orm.xml",
  mode: "initialLoad",
  maps: 'MYSCHEMA.INV,invCache'
)
INPUT FROM DataStream;

END APPLICATION InitialLoadOra2HC;

Once InitialLoadOra2HC has copied all the data, the following application will perform continuous replication of new data:

CREATE APPLICATION ReplicateOra2HC;

CREATE SOURCE OracleCDCSource USING OracleReader (
  Username:'striim',
  Password:'passwd',
  ConnectionURL:'203.0.113.49:1521:orcl',
  Tables:'MYSCHEMA.INVmyschema.ATM'
)
OUTPUT TO DataStream;

CREATE TARGET HazelOut USING HazelcastWriter (
  ConnectionURL: '203.0.1113.50:5702',
  ormFile:"Samples/Ora2HCast/invObject_orm.xml",
  mode: "incremental",
  maps: 'MYSCHEMA.INV,invCache'
)
INPUT FROM DataStream ;

END APPLICATION ReplicateOra2HCInitialLoadOra2HC;

Striim provides a wizard for creating applications that read from Oracle and write to HBase. See Creating an application using a wizard for details.

The following sample application will continuously replicate changes to MYSCHEMA.MYTABLE to the HBase Writer table mytable in the column family oracle_data:

CREATE SOURCE OracleCDCSource USING OracleReader ( 
  Username:'striim',
  Password:'passwd',
  ConnectionURL:'203.0.113.49:1521:orcl',
  Tables:'MYSCHEMA.MYTABLE',
  ReaderType: 'LogMiner',
  CommittedTransactions: false
)
INSERT INTO DataStream;

CREATE TARGET HBaseTarget USING HBaseWriter(
  HBaseConfigurationPath:"/usr/local/HBase/conf/hbase-site.xml",
  Tables: 'MYSCHEMA.MYTABLE,mytable.oracle_data'
INPUT FROM DataStream;

Notes:

The Tables value may map Oracle tables to HBase tables and column families in various ways:

The following sample application uses data from OracleReader, but you can do the same thing with any of the other CDC readers.

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),
  PRIMARY KEY (MERCHANT_ID));
COMMIT;

CREATE SOURCE OracleCDCSource USING OracleReader (
  StartTimestamp:'07-OCT-2015 18:37:55',
  Username:'qatest',
  Password:'qatest',
  ConnectionURL:'192.0.2.0:1521:orcl',
  Tables:'QATEST.POSAUTHORIZATIONS',
  OnlineCatalog:true,
  FetchSize:1,
  Compression:true
) 
OUTPUT TO DataStream;

CREATE TARGET HiveTarget USING HDFSWriter(
	filename:'ora_hive_pos.bin',
	hadoopurl:'hdfs://localhost:9000/output/’,
)
FORMAT USING AvroFormatter (
	schemaFileName: ’ora_hive_pos.avsc'
)
INPUT FROM DataStream;

{
  "namespace": "waevent.avro",
  "type" : "record",
  "name": "WAEvent_Record",
  "fields": [
    {
    "name" : "data",
    "type" : { "type": "map","values":"string" }
     },
    {
    "name" : "before",
    "type" : ["null",{"type": "map","values":"string" }]
    },
    {
     "name" : "metadata",
     "type" : { "type": "map","values":"string" }
    }
  ]
}

CREATE TABLE OracleHive
ROW FORMAT SERDE'org.apache.hadoop.hive.serde2.avro.AvroSerDe' 
STORED AS INPUTFORMAT'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'
OUTPUTFORMAT'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
TBLPROPERTIES ('avro.schema.url'='hdfs://localhost:9000/avro/ora_hive_pos.avsc');

hive> describe formatted oraclehive;
OK
# col_name            	data_type           	comment             
	 	 
data                	map<string,string>  	                    
before              	map<string,string>  	                    
metadata            	map<string,string>  	                    
	 	 
….                 
Time taken: 0.481 seconds, Fetched: 34 row(s)
hive> 

hive>LOAD DATA INPATH '/output/ora_hive_pos.bin' OVERWRITE INTO TABLE OracleHive;

INSERT INTO POSAUTHORIZATIONS VALUES('COMPANY 1',
'D6RJPwyuLXoLqQRQcOcouJ26KGxJSf6hgbu',6705362103919221351,0,'20130309113025','0916',
'USD',2.20,5150279519809946,41363,'Quicksand');
INSERT INTO POSAUTHORIZATIONS VALUES('COMPANY 2',
'OFp6pKTMg26n1iiFY00M9uSqh9ZfMxMBRf1',4710011837121304048,4,'20130309113025','0815',
'USD',22.78,5985180438915120,16950,'Westfield');
INSERT INTO POSAUTHORIZATIONS VALUES('COMPANY 3',
'ljh71ujKshzWNfXMdQyN8O7vaNHlmPCCnAx',2553303262790204445,6,'20130309113025','0316',
'USD',218.57,0663011190577329,18224,'Freeland');
INSERT INTO POSAUTHORIZATIONS VALUES('COMPANY 4',
'FZXC0wg0LvaJ6atJJx2a9vnfSFj4QhlOgbU',2345502971501633006,3,'20130309113025','0813',
'USD',18.31,4959093407575064,55470,'Minneapolis');
INSERT INTO POSAUTHORIZATIONS VALUES('COMPANY 5',
'ljh71ujKshzWNfXMdQyN8O7vaNHlmPCCnAx',6388500771470313223,2,'20130309113025','0415',
'USD',314.94,7116826188355220,39194,'Yazoo City');
UPDATE POSAUTHORIZATIONS SET BUSINESS_NAME = 'COMPANY 1A' where pos= 0;
UPDATE POSAUTHORIZATIONS SET BUSINESS_NAME = 'COMPANY 5A' where pos= 2;
UPDATE POSAUTHORIZATIONS SET BUSINESS_NAME = 'COMPANY 4A' where pos= 3;
UPDATE POSAUTHORIZATIONS SET BUSINESS_NAME = 'COMPANY 2A' where pos= 4;
UPDATE POSAUTHORIZATIONS SET BUSINESS_NAME = 'COMPANY 3A' where pos= 6;
DELETE from POSAUTHORIZATIONS where pos=6;
COMMIT;

hive> select * from oraclehive;
OK
{"3":"0","2":"6705362103919221351","1":"D6RJPwyuLXoLqQRQcOcouJ26KGxJSf6hgbu",
"10":"Quicksand","0":"COMPANY 1","6":"USD","5":"0916","4":"20130309113025","9":"41363",
"8":"5150279519809946"} NULL    {"TxnID":"10.23.1524","RbaSqn":"209",
"TableSpace":"USERS","CURRENTSCN":"1939875","OperationName":"INSERT",
"ParentTxnID":"10.23.1524","SegmentType":"TABLE","SessionInfo":"UNKNOWN",
"ParentTxn":"QATEST","Session":"143","BytesProcessed":"760",
"TransactionName":"","STARTSCN":"","SegmentName":"POSAUTHORIZATIONS","COMMITSCN":"",
"SEQUENCE":"1","RbaBlk":"57439","ThreadID":"1","SCN":"193987500000588282738968494240000",
"AuditSessionId":"73401","ROWID":"AAAXlEAAEAAAALGAAA",
"TimeStamp":"2015-10-08T14:58:55.000-07:00","Serial":"685",
"RecordSetID":" 0x0000d1.0000e05f.0010 ","TableName":"QATEST.POSAUTHORIZATIONS",
"SQLRedoLength":"325","Rollback":"0"}
{"3":"4","2":"4710011837121304048","1":"OFp6pKTMg26n1iiFY00M9uSqh9ZfMxMBRf1",
"10":"Westfield","0":"COMPANY 2","6":"USD","5":"0815","4":"20130309113025","9":"16950",
"8":"5985180438915120"} NULL    {"TxnID":"10.23.1524","RbaSqn":"209",
"TableSpace":"USERS","CURRENTSCN":"1939876","OperationName":"INSERT",
"ParentTxnID":"10.23.1524","SegmentType":"TABLE","SessionInfo":"UNKNOWN",
"ParentTxn":"QATEST","Session":"143","BytesProcessed":"762",
"TransactionName":"","STARTSCN":"","SegmentName":"POSAUTHORIZATIONS","COMMITSCN":"",
"SEQUENCE":"1","RbaBlk":"57441","ThreadID":"1","SCN":"193987600000588282738969804960001",
"AuditSessionId":"73401","ROWID":"AAAXlEAAEAAAALGAAB",
"TimeStamp":"2015-10-08T14:58:56.000-07:00","Serial":"685",
"RecordSetID":" 0x0000d1.0000e061.0010 ","TableName":"QATEST.POSAUTHORIZATIONS",
"SQLRedoLength":"327","Rollback":"0"}
...
Time taken: 0.238 seconds, Fetched: 11 row(s)

hive> select metadata["TimeStamp"], metadata["TxnID"], metadata["TableName"],
data from orclehive where metadata["OperationName"]="UPDATE";
2015-10-08T14:58:56.000-07:00	5.26.1740	QATEST.POSAUTHORIZATIONS
	{"0":"COMPANY 1A","1":"D6RJPwyuLXoLqQRQcOcouJ26KGxJSf6hgbu"}
2015-10-08T14:58:56.000-07:00	5.26.1740	QATEST.POSAUTHORIZATIONS
	{"0":"COMPANY 2A","1":"OFp6pKTMg26n1iiFY00M9uSqh9ZfMxMBRf1"}
...
Time taken: 0.088 seconds, Fetched: 5 row(s)

See Hive Writer for information on storage types supported and limitations.

The following example assumes the following Oracle source table:

create table employee (Employee_name varchar2(30), 
Employeed_id number, 
CONSTRAINT employee_pk PRIMARY KEY (Employeed_id));

and the following Hive target table:

CREATE TABLE employee (emp_name string, emp_id int)
CLUSTERED BY (emp_id) into 2 buckets 
STORED AS ORC TBLPROPERTIES ('transactional'='true');

The following application will load existing data from Oracle to Hive:

CREATE SOURCE OracleJDBCSource USING DatabaseReader (
  Username:'oracleuser',
  Password:'********',
  ConnectionURL:'192.0.2.75:1521:orcl',
  Tables:'DEMO.EMPLOYEE',
  FetchSize:1
)
OUTPUT TO DataStream;

CREATE TARGET HiveTarget USING HiveWriter (
  ConnectionURL:’jdbc:hive2://localhost:10000’,
  Username:’hiveuser’, 
  Password:’********’,
  hadoopurl:'hdfs://18.144.17.75:9000/',
  Mode:’initiaload’,
  Tables:’DEMO.EMPLOYEE,employee’
)
INPUT FROM DataStream;

Once initial load is complete, the following application will read new data and continuously replicate it to Hive:

CREATE SOURCE OracleCDCSource USING OracleReader (
  Username:'oracleuser',
  Password:'********',
  ConnectionURL:'192.0.2.75:1521:orcl',
  Tables:'DEMO.EMPLOYEE',
  FetchSize:1
)
OUTPUT TO DataStream;

CREATE TARGET HiveTarget USING HiveWriter (
  ConnectionURL:’jdbc:hive2://192.0.2.76:10000’,
  Username:’hiveuser’, 
  Password:’********’,
  hadoopurl:'hdfs://192.0.2.76:9000/',
  Mode:’incrementalload’,
  Tables:’DEMO.EMPLOYEE,employee keycolumns(emp_id)’’
)
INPUT FROM DataStream;

Before following these instructions:

You will need the following information to complete the wizard:

Note that FetchSize: 1 is appropriate for development, but should be increased in a production environment. See Oracle Reader properties for more information.

DatabaseWriter can continuously replicate one or many Oracle tables to a SAP HANA database. First, create a table in SAP HANA corresponding to each Oracle table to be replicated. Then load the existing data using DatabaseReader. For example, to replcate all tables in MYORADB to MYSAPDB:

CREATE SOURCE OracleJDBCSource USING DatabaseReader (
  Username:'Striim',
  Password:'****',
  ConnectionURL:'jdbc:oracle:thin:@203.0.113.49:1521:orcl',
  Tables:'MYORADB.%'
)
OUTPUT TO DataStream;

CREATE TARGET SAPHANAInitialLoad USING DatabaseWriter (
  ConnectionURL:'jdbc:sap://203.0.113.50:39013/?databaseName=MYASPDB&currentSchema=striim',
  Username:'striim',
  Password:'******',
  Tables:'MYORADB.%,MYSAPDB.%'
)
INPUT FROM DataStream; 

See Database Reader and Database Writer for more information about the properties.Database Writer

Once the initial load is complete, the following application will continuously replicate new data from Oracle to SAP HANA:

CREATE SOURCE OracleCDCSource USING OracleReader ( 
  Username: 'Striim',
  Password: '******',
  ConnectionURL: '203.0.113.49:1521:orcl',
  Compression:'True',
  Tables: 'MYORADB.%'
 ) 
OUTPUT TO DataStream;

CREATE TARGET SAPHANAContinuous USING DatabaseWriter(
  ConnectionURL:'jdbc:cassandra://203.0.113.50:9042/mykeyspace',
  Username:'striim',
  Password:'******',
  Tables: 'MYORADB.%,MYSAPDB.% '
INPUT FROM DataStream;

When the input stream of a SAP HANA DatabaseWriter target is the output of an Oracle source (Database Reader, Incremental Batch Reader, or Oracle Reader), the following types are supported:

Striim provides a wizard for creating applications that read from Oracle and write to Snowflake. See Creating an application using a wizard for details.

SnowflakeWriter can continuously replicate one or many Oracle tables to Snowflake. First, create a table in Snowflake corresponding to each Oracle table to be replicated. Then load the existing data using DatabaseReader, for example:

CREATE SOURCE OracleJDBCSource USING DatabaseReader (
  Username: 'striim',
  Password: '******',
  ConnectionURL: 'jdbc:oracle:thin:@//127.0.0.1:1521/xe',
  Tables: 'QATEST.%'
OUTPUT TO DataStream;

CREATE TARGET SnowflakeInitialLoad USING SnowflakeWriter (
  ConnectionURL: 'jdbc:snowflake://hx75070.snowflakecomputing.com/?db=DEMO_DB&schema=public',
  username: 'striim',
  password: '******',
  Tables: 'QATEST.%,DEMO_DB.PUBLIC.%',
  appendOnly: true
)
INPUT FROM DataStream;

Once the initial load is complete, the following application will read new data using LogMiner and continuously replicate it to Snowflake:

CREATE SOURCE OracleCDCSource USING OracleReader (
  Username: 'striim',
  Password: '******',
  ConnectionURL: 'jdbc:oracle:thin:@//127.0.0.1:1521/xe',
  Tables: 'QATEST.%'
OUTPUT TO DataStream;

CREATE TARGET SnowflakeCDC USING SnowflakeWriter (
  ConnectionURL: 'jdbc:snowflake://hx75070.snowflakecomputing.com/?db=DEMO_DB&schema=public',
  username: 'striim',
  password: '******',
  Tables: 'QATEST.%,DEMO_DB.PUBLIC.%'  
)
INPUT FROM DataStream;

For for more information, see Snowflake Writer.