Startup
Business
Enterprise
On-Premise
Add-on

Overview

Change data capture (CDC) is an approach to data integration that is based on the identification, capture, and delivery of the changes made to the source database and stored in the database redo log (also called transaction log).

The changes are captured without making application-level changes and without having to scan transactional tables. CDC is a perfect solution for non-intrusive, low-impact real-time data integration.

The Etlworks Integrator supports native log-based change data capture for PostgreSQL, SQL Server, MySQL, Oracle, DB2, and MongoDB.

We are using a heavily modified embedded Debezium engine for CDC.

Read about other change replication techniques available in the Etlworks Integrator.

Supported CDC topologies

Cloud, with optional SSH tunnel

Read more about using SSH tunnel to connect to the on-premise database.

Hybrid-cloud with data integration agents

Read more about data integration agents.

On-premise

Enable change data capture for the source database

Enabling CDC is different for each database. Please use the following tutorials:

CDC connectors

Read about available CDC connectors.

Parameters available for specific database connectors:

CDC pipeline

The end-to-end CDC pipeline extracts data from a source CDC-enabled database and loads it into any supported destination:

another relational database
data warehouse such as Snowflake, Redshift, Google BigQuery, Azure Synapse, Greenplum, Vertica
cloud storage system or data lake
a file storage system
a message queue such as Kafka, Kinesis, etc.

Built-in CDC pipelines

Etlworks includes the following built-in CDC pipelines:

CDC Extract

There are two generic CDC flow types that can be used together with other flows to CDC data from the source to the destination.

There are also specialized CDC flow types for specific destinations, but the extract part remains the same.

Stream CDC events, create files in the local or cloud storage

In this scenario, the CDC events (INSERTS, UPDATES, and DELETES) are extracted from the transaction log of the source database and stored as CSV and/or JSON files in the local or cloud storage.

This Flow type does not require any additional elements of the infrastructure, such as the message queue.

The separate load Flow reads the files, then transforms and loads data into the destination. The load Flow deletes the processed files. Read about available CDC pipelines.

There is no need to create a separate Flow for the initial load. The first time it connects to a CDC-enabled source database, it reads a consistent snapshot of all of the included databases and tables. When that snapshot is complete, the Flow continuously reads the changes that were committed to the transaction log and generates the corresponding insert, update, and delete events.

Create CDC connection and Flow

Step 1. Create a CDC Connection for the source database. Configure the following parameters:

Enable Serialize CDC events as CSV files. If you wish to create JSON files, you can also enable Serialize CDC events as JSON files. These options can be enabled or disabled independently from each other.
Include Database(s): the comma-separated list of databases(s) to poll CDC events from.
Include Table(s): the command-separated list of fully qualified table names or regular expressions to match the table names. You can override this in the FROM attribute of the source-to-destination transformation. If you are planning to read data from tens or hundreds of tables, consider using regular expressions or {tokenizing} the Included Table(s) or FROM.

Step 2. Create a new Flow using the Flow type Stream CDC events, create files.

Step 3. Create a CDC events connection if you wish to create CDC events in the local storage. Create one of the following cloud storage connections if you want to create CDC events in the cloud storage or data lake:

Add source-to-destination transformation

Step 4. Add a single source-to-destination transformation and set the following parameters:

Source Connection: the CDC Connection created in step 1
FROM: the following options are available:
- *: the system will get the list of the included tables from the CDC Connection
- a regular expression to match the table names
- a comma-separated list of fully qualified table names or regular expressions matching the table names.
- {token}

You can select tables from the list of available tables:

Example with a comma-separated list of included tables:

CDC key: [db]_[table]_cdc_stream
Include Table(s): test.inventory, test.payment, test.customer
FROM: *
The following files will be created in the local storage:
- {app.data}/debezium_data/events/test_inventory_cdc_stream_uuid.csv
- {app.data}/debezium_data/events/test_payment_cdc_stream_uuid.csv
- {app.data}/debezium_data/events/test_customer_cdc_stream_uuid.csv

Example with a regular expression:

CDC key: [db]_[table]_cdc_stream
Include Table(s): ^[t][e][s][t][.][aA][a-zA-Z\d_\s]+
FROM: *
The following files will be created in the local storage:
- {app.data}/debezium_data/events/test_inventory_cdc_stream_uuid.csv
- {app.data}/debezium_data/events/test_payment_cdc_stream_uuid.csv
- {app.data}/debezium_data/events/test_customer_cdc_stream_uuid.csv

The {app.data}/debezium_data is a default root location for the CDC events serialized as CSV and JSON files. You can override it by changing the value of the property Location of the CDC events under the CDC Connection > Storage.

The better solution is to override the destination in the transformation (see below).

Optionally set destination connection, override format, and CDC Key

Step 5. Select the connection created in step 3, which can be one of the following:

Optionally set Format, and override the CDC Key in the TO.

The right part of the source-to-destination transformation overrides the following parameters of the CDC connection:

The Destination Connection overrides the storage type, the location of the CDC files, the storage credentials, and whether the files should be gzipped set in the CDC connection under the section Storage.
The Destination Format overrides the following parameters set in the CDC connection.

The TO overrides the CDC Key.

Schedule CDC flow

Step 5. Schedule the extract Flow. We recommend using a continuous run Schedule type. The idea is that the extract Flow runs indefinitely or until there are no CDC events to extract, or it stops if there is an error. When the Flow stops for any of the reasons above, it restarts automatically after the configured number of seconds.

Stream CDC events to a message queue

In this scenario, the CDC events (INSERTS, UPDATES, and DELETES) are extracted from the transaction log of the source database and sent to the message queue, such as Kafka and Azure Event Hubs.

Message queues such as Kafka and Azure Event Hubs support transactional polling, which makes the entire process very reliable and highly scalable. The only downside is the requirement to have a message queue, which is a separate element of the infrastructure that is not managed by the Etlworks Integrator.

Here is a list of fully managed message queue services compatible with Kafka consumer and producer APIs:

Azure Event Hubs: read how to create a Connection for Azure Event Hub.
Confluent Platform
CloudKafka: only service with a free forever tier. The free tier is very limited but can be used for testing.
Aiven for Apache Kafka
AWS managed Kafka (MSK): note that due to the limitation of the MSK, it is only possible to connect to MSK endpoints from the same VPC. Read more here.

Etlworks also supports streaming CDC events to the following destinations:

There is no need to create a separate Flow for the initial load. The first time it connects to a CDC-enabled source database, it reads a consistent snapshot of all of the included databases. When that snapshot is complete, the Flow continuously reads the changes that were committed to the transaction log and generates the corresponding INSERT, UPDATE, and DELETE events.

Step-by-step process

Step 1. Create a CDC Connection for the source database. Configure the following parameters:

Include Databases(s): the comma-separated list of databases(s) to extract CDC events from.
Include Table(s): the command-separated list of fully qualified table names or regular expressions to match the table names. You can override this in the FROM attribute of the source-to-destination transformation. If you are planning to read data from tens or hundreds of tables, consider using regular expressions or {tokenizing} the Whitelisted Table(s) or FROM.

Step 2. Create a Kafka or Azure Event Hub Connection. The other options are:

Step 3. Create JSON (recommended) or Avro Format.

Step 4. Start creating a Flow by selecting the Stream CDC event to message queue from the gallery.

Step 5. Add a single source-to-destination transformation where:

Source Connection: CDC Connection created in step 1.
FROM: the following options are available:
- *: the system will get the list of the included tables from the CDC Connection
- a regular expression to match the table names
- a comma-separated list of fully qualified table names or regular expressions matching the table names
- {token}

You can select tables from the list of the available tables:

Destination Connection: a message queue Connection created in step 2, for example
Kafka or Azure Event Hub.
Format: JSON or Avro Format created in step 3.
TO: any name, for example, topic.

Example:

CDC key: [db]_[table]_cdc_stream
Include Table(s): test.inventory, test.payment, test.customer
FROM: *

The following topics will be created and the Flow will send CDC events to these topics:

test_invemtory_cdc_stream
test_payment_cdc_stream
test_customer_cdc_stream

Step 6. Schedule the extract Flow. We recommend using a continuous run Schedule type. The idea is that the extract Flow runs indefinitely or until there are no CDC events to extract, or it stops if there is an error. When the Flow stops for any of the reasons above, it restarts automatically after the configured number of seconds.

Insert events from all tables into a single topic

If you wish to insert all events from all tables into a single topic in the message queue:

Hardcode the CDC Key. CDC Key = topic name.

Enable Update fields for each event.

Include attributes available in the original CDC event emitted by Debezium

A typical CDC event emitted by Debezium contains the following elements:

before: an optional field that specifies the state of the row before the event occurred.
after: an optional field that specifies the state of the row after the event occurred.
source: a mandatory field that describes the source metadata for the event.
op: a mandatory string that describes the type of operation that caused the connector to generate the event.
ts_ms: an optional field that displays the time at which the connector processed the event,
transaction: transaction metadata,
schema: the schema, which describes the structure of the payload.
fields: the JDBC schema.

By default, the connector only retains the state of the event after the update and converts it to the flat JSON. The flat JSON is then added to the message queue.

You can change the structure of the event added to the queue by enabling parts of the original event emitted by Debezium by modifying the property Structure of the CDC event.

This option is ignored if the connector creates files.

Generate partition key for Kafka and Events Hub

Kafka’s topics are divided into several partitions. While the topic is a logical concept in Kafka, a partition is the smallest storage unit that holds a subset of records owned by a topic. Each partition is a single log file where records are written to it in an append-only fashion.

The messages within a partition are ordered; messages across a topic are not guaranteed to be ordered.

The Partition Key is used to send records to the specific Kafka or Azure Event Hub partition.

If the Partition Key is not set, the record will be sent to the random partition, so we strongly recommend configuring it.

Example of the partition key: [table].

The same connector-specific [tokens] that can be used as a part of the CDC Key are available for the Partition Key.

SQL Server tokens

[db]: database name.
[schema]: schema name.
[table]: table name.
[op]: the CDC operation: c (create), d (delete), u (update).
[timestamp]: the timestamp of the event in ms.
[version]: the connector version.
[connector]: the connector class name.
[name]: the connector name.
[ts_ms]: the timestamp of the event in ms.
[snapshot]: true for a snapshot.
[change_lsn]: change lns.
[commit_lsn]: commit lns.
[event_serial_no]: event serial number.

MySQL tokens

[db]: database name.
[table]: table name.
[op]: the CDC operation: c (create), d (delete), u (update).
[timestamp]: the timestamp of the event in ms.
[version]: the connector version.
[connector]: the connector class name.
[name]: the connector name.
[ts_ms]: the timestamp of the event in ms.
[snapshot]: true for a snapshot.
[server_id]: the MySQL server ID.
[gtid]: the GTID.
[file]: the binlog filename.
[pos]: the position in the binlog.
[row]: the row number.

Oracle tokens

[source]: fully qualified event source.
[op]: the CDC operation: c (create), d (delete), u (update).
[timestamp]: the timestamp of the event in ms .
[version]: the connector version.
[connector]: the connector class name.
[name]: the connector name.
[ts_ms]: the timestamp of the event in ms.
[snapshot]: true for a snapshot.
[txId]: the transaction id.
[scn]: the scn.

Postgres tokens

[db: database name.
[schema]: schema name.
[table]: table name.
[op]: the CDC operation: c (create), d (delete), u (update).
[timestamp]: the timestamp of the event in ms.
[version]: the connector version.
[connector]: the connector class name.
[name]:the connector name.
[ts_ms]: the timestamp of the event in ms.
[snapshot]:true for a snapshot.
[txId]: the transaction ID.
[lsn]: the lsn.
[xmin]: the xmin.

DB2 tokens

[db: database name.
[schema]: schema name.
[table]: table name.
[op]: the CDC operation: c (create), d (delete), u (update).
[timestamp]: the timestamp of the event in ms.
[version]: the connector version.
[connector]: the connector class name.
[name]:the connector name.
[ts_ms]: the timestamp of the event in ms.
[snapshot]:true for a snapshot.
[change_lsn].
[commit_lsn].

MongoDB tokens

[db]: the database name.
[collection]: the collection name.
[op]: the CDC operation: c (create), d (delete), u (update).
[timestamp]: the timestamp of the event in ms.
[version]: the connector version.
[connector]: the connector class name.
[name]: the connector name.
[ts_ms]: the timestamp of the event in ms.
[snapshot]: true for a snapshot.
[order]: the order of the event.
[h]: the h of the event.

Stream CDC events from the message queue to any destination

The Flow described in this article streams CDC events stored in a message queue into any supported destination. Read more.

Database-specific use cases

By default, our CDC connectors are configured to support the most common scenarios. Below are some database-specific tips for less common use cases.

MySQL use cases

Legacy and Modern implementations

MySQL CDC connector includes two implementations:

Modern: Modern implementation is based on the common connector framework used by all CDC connectors. The Modern implementation supports adding new tables to the existing connection using an incremental snapshot mechanism.
Legacy: Legacy implementation is deprecated but can be used for backward compatibility with the CDC Flows created before we introduced the Modern implementation. The Legacy implementation supports adding new tables to the existing connection using a parallel snapshot mechanism (enabled by default) which is not as reliable as the incremental snapshot supported by the Modern implementation.

MySQL Snapshot modes

Snapshot mode specifies the criteria for running a snapshot upon startup of the connector. Read about MySQL snapshot modes.

Add new tables to the pipeline (MySQL)

Make sure the snapshot mode is set to one of the following:

ad-hoc initial.
ad-hoc schema only.
always_recover_schema.
schema_only_always_recover_schema.

If you want to trigger the incremental snapshot of the new table( s) make sure Automatically trigger ad-hoc snapshot for new tables is enabled. Also, make sure that the database.history.store.only.captured.tables.ddl is enabled (it is enabled by default).

Add new tables to the pipeline:

Stop the Flow.
Add new tables to the Included Tables field in the CDC connection or to the FROM field in the CDC flow. IMPORTANT: the table names must be specified in a fully-qualified format. Adding new tables using regular expressions is not supported.
Restart the Flow.

Capture data from compressed binlog

MySQL CDC connector supports capturing data from the compressed binlog. There is no extra configuration required.

Work with spatial (geometry) data types (MySQL)

By default, our MySQL CDC connector encodes spatial (geometry) data types using WKB format and then uses Base64 encoding on top of that. If you want the connector to convert the geometry data to human-readable JSON, enable Convert Geometry WKB to JSON.

Stream CDC events from the specific binlog file, position, and optionally GTID

It is possible to configure the MySQL CDC connector to start streaming from the specific binlog file and position.

It is only available when the connector is switching from the snapshot to streaming mode, and the snapshot mode is set to ad-hoc initial , ad-hoc schema only,initial ,schema_only , always_recover_schema or schema_only_always_recover_schema.

Enable cursor-based result set when performing an initial snapshot

By default, the MySQL CDC connector reads data from the large dataset using the recommended technique by creating the JDBC statement with the flags ResultSet#TYPE_FORWARD_ONLY and ResultSet#CONCUR_READ_ONLY, and by setting the fetch size hint to Integer#MIN_VALUE.

In some rare cases, it could cause the following error: "No statements may be issued when any streaming result sets are open and in use on a given connection".

The alternative to this approach is to use the useCursorFecth property. When this property is set to true the MySQL server creates a temp table from the actual table, then streams data from the temp table. It requires more resources to be allocated by the MySQL server but prevents the error above.

To enable a cursor-based result set simply add database.useCursorFetch=true to Other Parameters:

Postgres use cases

Postgres Snapshot modes

Snapshot mode specifies the criteria for running a snapshot upon startup of the connector. Read about Postgres snapshot modes.

Add new tables to the pipeline (Postgres)

If you want to trigger the incremental snapshot of the new table( s) make sure Automatically trigger ad-hoc snapshot for new tables is enabled.

Add new tables to the pipeline:

Stop the Flow.
Add new tables to the Included Tables field in the CDC connection or to the FROM field in the CDC flow. IMPORTANT: the table names must be specified in a fully-qualified format. Adding new tables using regular expressions is not supported.
Restart the Flow.

Avoid abnormal consumption of WAL database disk space

In certain cases, it is possible that PostgreSQL disk space consumed by WAL files either experiences spikes or increases out of usual proportions.

This article suggests a solution for the problem.

Here is an example of the configuration which a) makes a connector to emit an empty heartbeat event every minute, and b) makes a connector to UPSERT a record into the heartbeat table on each heartbeat event.

Propagate default values

If option Propagate default values is enabled and a default value is specified for a column in the database schema, the PostgreSQL connector will attempt to propagate this value to the stored schema whenever possible.

This option is disabled by default.

SQL Server use cases

SQL Server Snapshot modes

Snapshot mode specifies the criteria for running a snapshot upon startup of the connector. Read about SQL Server snapshot modes.

Add new tables to the pipeline (SQL Server)

If you want to trigger the incremental snapshot of the new table( s) make sure Automatically trigger ad-hoc snapshot for new tables is enabled.

Add new tables to the pipeline:

Stop the Flow.
Add new tables to the Included Tables field in the CDC connection or to the FROM field in the CDC flow. IMPORTANT: the table names must be specified in a fully-qualified format. Adding new tables using regular expressions is not supported.
Restart the Flow.

Work with binary and spatial (geometry) data types (SQL Server)

It is recommended to serialize columns with binary and spatial (geometry) data as hex-encoded strings by setting property binary.handling.mode to hex.

Oracle use cases

Connect to read-only Oracle database

To stream data from a read-only database set CDC Adapterset tologminer and enable Run connector in read-only mode. When this option is enabled the connector will not attempt to flush the LGWR buffer to disk, allowing connecting to read-only databases.

Use JDBC URL to configure CDC connection

Due to the variety of URL formats supported by the Oracle JDBC driver, it is recommended to use the fully qualified Oracle JDBC URL to configure the CDC connection. You still need to enter the Host the Port and the Database, also (optionally) the PDB name.

Oracle Snapshot modes

Snapshot mode specifies the criteria for running a snapshot upon startup of the connector. Read about Oracle snapshot modes.

Improve the performance of the snapshot

When the Oracle CDC connector starts a snapshot, it reads the current system change number (SCN) position from the server’s redo log. It then scans all of the relevant database tables and schemas as valid at the SCN position that was read in the previous step (SELECT * FROM … AS OF SCN 123) and generates a READ event for each row. Flashback queries (AS OF) can be slow because Oracle needs to 'play' the DML events (insert/update/delete) with reference to a specific point in time.

Enable Disable Flashback Query (AS OF) when creating snapshot to disable the flashback queries during the snapshot. This could improve the performance of the snapshot in some environments.

Add new tables to the pipeline (Oracle)

Make sure the snapshot mode is set to one of the following:

ad-hoc initial.
ad-hoc schema only.
always_recover_schema.
schema_only_always_recover_schema.

Add new tables to the pipeline:

Stop the Flow.
Add new tables to the Included Tables field in the CDC connection or to the FROM field in the CDC flow. IMPORTANT: the table names must be specified in a fully-qualified format. Adding new tables using regular expressions is not supported.
Restart the Flow.

DB2 use cases

DB2 Server Snapshot modes

Snapshot mode specifies the criteria for running a snapshot upon startup of the connector. Read about DB2 snapshot modes.

Add new tables to the pipeline (DB2)

Add new tables to the pipeline:

Stop the Flow.
Add new tables to the Included Tables field in the CDC connection or to the FROM field in the CDC flow. IMPORTANT: the table names must be specified in a fully-qualified format. Adding new tables using regular expressions is not supported.
Restart the Flow.

MongoDB use cases

Use Cluster URL to browse data

Technically speaking the Cluster URL is not used by the CDC connector. It is, however, required if you want to browse the data in MongoDB using the Etlworks Explorer.

Cluster URL

Browsing data in Explorer

Connect to sharded MongoDB cluster

By default, our MongoDB CDC connector always attempts to connect to the primary node in the sharded cluster. If the node is down, the connector will try to reconnect to the newly elected primary node.

Two parameters control this behavior:

Ensure that there is an active session on the replica set: if this option is enabled, the connector will ensure that there is an active session on the replica set mongo client before determining the primary node. If this option is disabled, the connector will connect to the first node in the cluster returned by the topology discovering service. It is not guaranteed that this node will be primary. This option is enabled by default.
Use secondary node if primary is not found: if this option is enabled (and the option above is also enabled) and the connector is unable to determine the primary node, it will attempt to connect to the secondary node. This option is enabled by default.

Add new collections to the pipeline

To add new collections to the pipeline:

Stop the Flow.
Add new collections to the Included Collections list.
Restart the Flow.

The Flow will start steaming changes for new collections right away.

The MongoDB connector does not currently support snapshotting of the new collections, so if you need it you will have to create a new CDC Connection specifically for these collections.

Enable support for MongoDB Change Streams

Change streams allow applications to access real-time data changes without the complexity and risk of tailing the oplog. Applications can use change streams to subscribe to all data changes on a single collection, a database, or an entire deployment and immediately react to them. Read more about MongoDB Change Streams.

Etlworks MongoDB CDC connector supports 3 capture modes:

oplog: enable to capture changes from oplog.
change_streams: enable to capture via MongoDB Change Streams mechanism; update message does not contain the full message.
change_streams_update_full: enable to capture via MongoDB Change Streams mechanism, update message contains the full message.

Serialize CDC events as JSON documents

When creating a CDC connection for MongoDB it is recommned to configure it to create JSON documents which preserve the original structure of the source documnet but flatten columns stored using extended JSON (typically ids and timestamp columns).

Step 1. Disable Serialize CDC events as CSV files, enable Serialize CDC events as JSON files and disable Close CSV file if header has changed.

Step 2. Enable all 3 parameters below:

Enabling all 3 parameters will create a JSON array where each record is exacly the same as the record in the source except flattened columns stored using extended JSON. JSON files can be easely loaded as is into the destination database, for example Snowflake, or just stored in the cloud data storage or data lake.

Ad-hoc and incremental snapshots

Ad-hoc snapshots

When the initial snapshot is completed, it is not executed again as further data are acquired via streaming.

Sometimes it would be really useful to re-execute a snapshot, either completely or for only a subset of tables. Examples of such use cases include:

a new table was added to the list of captured tables.
some of the topics in the message queue were deleted, and their content needs to be rebuilt.
the existing table needs to be re-snapshotted for whatever reason.

Incremental snapshot

Initial snapshots are a great way to obtain a consistent set of all data stored in a database upfront. But there are also some downsides to this approach, though:

For larger datasets, it can take very long to complete, hours or even days.
The snapshot must be completed before the start of streaming.
Snapshots are not resumable If a snapshot hasn’t been completed when the connector gets stopped or restarted, the snapshot must be restarted from scratch.
New tables that are added among the captured ones later are not snapshotted.

To mitigate these issues, a new mechanism called incremental snapshotting has been introduced. The table is snapshotted in chunks, and streaming and snapshot events are blended together.

The table must have a primary key in order for the connector to be able to execute an incremental snapshot for that table.

Trigger incremental snapshot of the new tables

The incremental snapshot of the new tables requires restarting the CDC extract Flow

Make sure Automatically trigger ad-hoc snapshot for new tables is enabled.

Make sure the snapshot mode is set to one of the following:

ad-hoc initial.
ad-hoc.
ad-hoc schema only.
always_recover_schema.
schema_only_always_recover_schema.

Add new tables to the pipeline:

Stop the Flow.
Add new tables to the Included Tables field in the CDC connection or to the FROM field in the CDC flow. IMPORTANT: the table names must be specified in a fully-qualified format. Adding new tables using regular expressions is not supported.
Restart the Flow.

Trigger ad-hoc snapshot of the existing tables

The ad-hoc snapshot of the existing tables requires restarting the CDC extract Flow.

How ad-hoc snapshot works

It is assumed that there is a configurable data collection that contains the list of database tables to snapshot. We call it a signal. The signal can be one of the following:

A database table. The database doesn't have to be CDC-enabled, but the Flow should be able to delete records from a signal table.
A file in any of the supported file storage systems. The flow must be able to delete the file.

The signal table is used as a container to list all tables which you want to re-snapshot. It is not the actual table to re-snapshot.

The CDC flow monitors the signal. When changes are detected, the Flow reads the list of tables in the signaland triggers the ad-hoc for the tables listed in the data collection.

The flow immediately deletes the processed signal:

If the signal is a database table, the Flow deletes all records from that table.
If the signal is a file the Flow deletes the file.

The format of the signal

The signaltable or file must contain a single column that can have any name.

The tables to snapshot can be stored:

a) Horizontally - multiple comma-separated tables in a single row, for example test.inventory, test.payment, test.customer

b) Vertically - multiple rows with table names; each row can contain one or multiple comma-separated tables.

Configure ad-hoc snapshot

In Connections

Step 1. Create file storage or database connection. The connection must be read/write.

Step 2. For the file, storage connections create a format.

In CDC Flow.

Step 3. Add the connection and format created in steps 1-2 under the Connection tab of the CDC Flow.

In CDC connection.

Step 4. Make sure the snapshot mode is set to one of the following:

ad-hoc initial.
ad-hoc.
ad-hoc schema only.

Step 5. Configure the name of the signal data collection.

If the signal is a database table enter a fully qualified table name.

If the signal is a file enter the file name with extension but without the path.

Trigger add-hoc snapshot

Step 1. Stop CDC extract flow.

Step 2. To trigger the ad-hoc snapshot for the tables listed in the signal data collection, insert records into the signal table or create a signal file.

Step 3. Start CDC extract flow.

Tips and tricks

Configure data exchange formats

Most CDC flows (except dirrect CDC to database and CDC to message queue) create either CSV or JSON files (or both), then load them into the designated database or cloud data storage/data lake. There are several configuration options available for the CDC connection which control the format of the created files.

Convert NULL values in the source database into empty ('') strings in CSV

To convert NULL values in the source database into empty ('') strings in CSV enable Convert null to empty for the source CDC connection.

Do not enclose null values in double quotes

By default CDC connector encloses all values (including nulls) into the double quotes when creatting CSV files. Some databases, for exmaple Greenplum cannot correcly load empty strings ("") into nullable columns.

To stop enclosing null values in double quotes enable Do not enclose null in double quotes for the CDC connection.

Convert boolean true/false into numeric 1/0

Some databases do not have native boolean datatype and use numeric columns to store boolean values as 1 or 0.

To atomatically convert boolean (true/false) values in the source database into numerical 1/0 enable Convert boolean true/false into numerical 1/0 for the CDC connection.

Preserve end-of-line (EOL) characters in column's value

Some database does not allow loading CSV files with when column's value includes EOL characters, so the CDC connection automaticaly removes EOLs.

To preserve EOLs disable Remove EOL characres for the CDC connection.

Create CSV and JSON files using the specified character encoding

If the source database has tables with data encoded using any character set other than ASCII (for example UTF-8), you can configure the connector to create files using that specific encoding.

To preserve the encoding when loading files into the databases, you will need to set the same character encoding for the CSV Format.

If the encoding is not specified or set to no encoding, the JSON files are created with UTF-8 encoding.

Handle NULL and empty ('') values when loading CSV files into databases

Most databases and cloud data warehouses allow batch-loading CSV files created by CDC flow. Databases handle loading NULL values diffrently.

NULL is the same as empty ('')

Step 1. When creating a CDC Connection, enable Convert null to empty. Note that it is disabled by default.

Step 2. When creating a CVS Format for FROM and TO:

Disable Convert 'null' to null.
Enable Convert empty string to null.
Set Value for nullempty.

Loading empty values as NULL into Snowflake

When creating a Flow to load data into Snowflake, set String used to covert to and from SQL NULL to ('NULL','null','').

Loading empty values as NULL into Redshift

When creating a Flow to load data into Redshift, enable Load empty chars as null.

Loading empty values as NULL into Greenplum

For CDC connection, enable Do not enclose NULL in double quotes.

NULL is not the same as empty ('')

Step 1. When creating a CDC Connection, make sure Convert null to empty is disabled. Note that it is disabled by default.

Step 2. When creating a CVS Format for FROM and TO:

Enable Convert 'null' to null.
Disable Convert empty string to null.
Set Value for null to 'null'.

Loading not empty values as NULL into Snowflake

When creating a Flow to load files into Snowflake, set String used to convert to and from SQL NULL to ('NULL','null').

Working with JSON columns in the source database

When the source database stores data in the nested data structures (for example MongoDB) or a record includes column(s) with nested data elements serialized as JSONs (for example JSON columns in MySQL) the CDC connecctor can be configured to stringify JSON columns.

To convert JSON columns in the source to strings when creating CSV or/and JSON files enable all 3 parameters below for the CDC connection.

When CDC connector creates JSON files

Enabling all 3 options will create a JSON array where each record is exacly the same as the record in the source. JSON files can be easely loaded as is into the destination database, for example Snowflake, or just stored in the cloud data storage or data lake.

When CDC connector creates CSV files

Assuming that the source table has JSON columns, the values of the columns will be stringified preserving the original structure of the JSON columns. Stringified JSON columns in the CSV file can be easely loaded into the JSON columns in the destination database, for example Snowflake.

Handle source table DDL changes

If you are using a Flow that stores CDC events as CSV files and are expecting that the structure of the source table can be modified when the Flow is still streaming data from this table when creating the CDC Connection, enable Close CSV file if header has changed.

Preserve the exact data types of the columns in the source table when creating a destination table

The Etlworks Integrator can automatically create a destination table when loading CSV files. By default, the Flow will sample the file and set the data types of the columns in the destination table. This mechanism is not 100% accurate so the destination table might end up with columns that have different data types compared to the source table.

To prevent this from happening:

Step 1. When creating the CDC Connection, enable Save Metadata.

Step 2. When creating a CVS Format for FROM and TO.

Enable Save Metadata.
Keep All fields are strings disabled.

In addition, when enabled, this feature allows the CDC load Flow to automatically set the Lookup Fields required for the CDC MERGE by extracting the information about the primary keys from the source database. It significantly enhances the accuracy of the Predict Lookup Fields feature. Basically makes it bulletproof and very low cost.

Capture NOT NULL constraints

When Save Metadata is enabled the CDC connector captures the following attributes of the source columns: name, SQL data type, JDBC type, and size. It does not capture the NOT NULLconstraints by default. To configure the connector to capture NOT NULLconstraints enable this configuration option:

Create all columns as strings (VARCHAR)

The Etlworks Integrator can automatically create a destination table when loading CSV files. If you wish to create all columns as strings (VARCHAR), then:

Step 1. When creating the CDC Connection disable Save Metadata.

Step 2. When creating a CVS Format for FROM and TO:

Disable Save Metadata.
Enable All fields are strings.

Implement soft deletes

By default, Load Flow creates, updates, and deletes records in the destination table based on the type of the CDC event: c for create, u for update, and d for delete.

If you need to implement the soft delete - add the Extra columns with the specific function to the CDC Connection. The following functions are available:

cdc_boolean_soft_delete: the boolean column with a user-defined name will be added to the CDC stream. Values returned by the function: true for delete, false for all other events.
cdc_boolean_soft_delete(yyyy-MM-dd HH:mm:ss.SSS: the timestamp column with a user-defined name will be added to the CDC stream. Values returned by the function: timestamp for delete, nullfor all other events.

In this example, the column delete_at will be added to the CDC stream and its value will be set to the current timestamp with a milliseconds precision.

If any of the functions above are used the CDC event type is automatically switched to u (update).

Add columns to the CDC stream

By default, the CDC stream contains all columns in the source table. You can add columns with user-defined names to the stream using the Extra columns property of the CDC Connection.

Format of the property: column_name[=function[(parameter)],column_name[=function[(parameter)]

Available functions:

cdc_op: the CDC event type: c for create, u for update, and dfor delete
cdc_key: the CDC key in the user-defined Format
cdc_timestamp(yyyy-MM-dd HH:mm:ss.SSS): the timestamp of the CDC event. It is not guaranteed that this value will be unique for each event in the sequence of events.
cdc_event_seq: the unique 'number' of the event. This number grows sequentially.
cdc_boolean_soft_delete: the boolean column: truefor delete, false for all other events
cdc_timestamp_soft_delete(yyyy-MM-dd HH:mm:ss.SSS): the timestamp column: timestamp for delete, null for all other events.

Change CDC key and CDC event

The CDC event represents a row (a record) in the CDC stream. The CDC key represents the unique identifier of the group of events. All events with the same CDC key are stored in the same file.

Use Preprocessor to change the CDC event and/or CDC key.

Available scripting languages

JavaScript

Available variables

event: the CDC event serialized as JsonNode.
key: CDC key.
source: CDC source serialized as JsonNode.
transactionId: the transaction id associated with the event if transaction markers are enabled.

Get and set the values of the column in the event

To get the value of the column, serialized as a string:

var val = event.get('columns_name').asText();

To set the value:

event.put('columns_name', string);

Possible return values

true/ false: if false the event will be skipped.
TypedKeyValue<Boolean, String>: the key is an equivalent of true/false above, the value is CDC key to replace the automatically calculated CDC key.

Example:

var newCdcKey= 'abc';

value = new TypedKeyValue(true, newCdcKey);

Transaction markers

Consider, for instance, an e-commerce application that manages purchase orders. In a relational database, the information representing such order will typically be spread across multiple tables, e.g. PURCHASE_ORDER, ORDER_LINE, and SHIPMENT_ADDRESS. To examine the complete state of a purchase order in the database itself, you’d run a JOIN query against the three tables.

Things get more challenging when looking at the change events produced by CDC connectors.

The situation can be addressed by leveraging the new transaction metadata supported by most CDC connectors. When enabled, separate transaction markers will be created at the beginning and end of each transaction.

To enable transaction markers, modify the CDC Connection by enabling the Provide Transaction Metadata parameter. Optionally enable Log Transaction Metadata to add metadata (such as names of the tables involved in the transaction, as well as the number of affected records for each table) to the flow log.

How to stop CDC Flow

By default, the CDC flow never stops automatically. It is a recommended configuration.

You can configure the CDC Connection to stop when there are no more new CDC events for an extended period of time. The CDC Flow will stop if any of the above reaches the configured threshold.

Number of retries before giving up: the number of retries before CDC flow stops if there are new CDC events.
Retry N minutes before giving up: the number of minutes to retry before CDC flow stops if there are new CDC events.
Always stop after N minutes: the number of minutes to run the connector before stopping. The connector will stop even if there are new records in the queue.

Configure automatic stop with caution. It takes time to start streaming the CDC events, so stopping the flow prematurely could lead to a situation when no CDC events are streamed.

If none of the parameters are set (default), the CDC flow will run forever until manually stopped or there is an error.

To stop CDC Flow manually, click Stop / Cancel.

How to reset CDC Flow

Etlworks CDC connectors store the history of DDL changes for the monitored database in the history file and the current position in the transaction log in the offset file.

Typical CDC extract flow starts by snapshotting the monitored tables (A) or starts from the oldest known position in the transaction (redo) log (B), then proceeds to stream changes in the source database (C). If the Flow is stopped and restarted, it resumes from the last recorded position in the transaction log.

Occasionally it is required to "reset" the flow to restart the process from scratch (A) or (B).

Process

Step 1. Create a new Offset and History connection.

Step 2. Stop CDC flow if it is running.

Step 3. Open the connection created in step 1 in Explorer and delete .dat file(s) associated with the CDC connection.

Step 4. Start CDC flow. Depending on the configuration, the Flow will either perform the full snapshot or will start from the oldest position in the transaction log.

Monitor CDC Flows

Our ETL engine records real-time metrics for each active CDC stream. The following metrics are available:

Last Update(timezone: the time stamp in the server timezone of the last checkpoint. The metrics are updated every 60 seconds.
Last Record Processed (time ago): the human-readable timestamp of the last processed CDC event.
Messages/sec: the number of messages processed in one second recorded during the last checkpoint interval.
Last Record Latency: the last recorded latency. The latency is the difference between a timestamp of the change event in the source database and a timestamp of when the change event was processed by the CDC connector.
Max Latency: the maximum recorded latency since the last restart of the CDC connector.
Min Latency: the minimum recorded latency since the last restart of the CDC connector.
Avg Latency: the average latency since the last restart of the CDC connector.
Total Records Processed: the total number of the change events processed since the last restart of the CDC connector.
Records Processed Since Last Update: the number of the change events processed since the last update. Note that the metrics are updated every 60 seconds.

Access CDC Metrics from UI

To access CDC metrics from UI go to Flows or Schedules and click Running next to the Flow or Schedule name.

Then click View Running Tasks.

The metrics are displayed at the bottom of the Running Tasks window.

You can refresh the metrics manually:

or enable Auto refresh.

Access CDC metrics using API

GET CDC metrics for the specific Flow

This API returns metrics for the specific CDC Flow.

Authentication

Before making a call to API, you must receive a JWT token from the authentication API.

Step 1. Use any user with the Administrator role to call an Etlworks Integrator authentication endpoint and receive an access token.

Step 2. Use the access token, received in Step 1, to call the Etlworks Integrator API endpoints. The access token must be submitted as a header parameter, as in: Authorization:Bearer access-token.

Access tokens are short-lived and self-expiring. An access token gives you access to the APIs for approximately 10 minutes. It is recommended that you refresh the access token before each call to the API.

The API endpoint parameters:

PATH: /etl/rest/v1/tasks/{audit_id}/?type=CDC%20stream

EXAMPLE: https://app.etlworks.com/etl/rest/v1/tasks/80610/?type=CDC%20stream

METHOD: GET
HEADER: Authorization:Bearer access-token
REQUEST BODY: none
REQUEST CONTENT TYPE: none

Audit ID

audit_id is a required URL parameter. To get the audit_id for the specific Flow, use the Flow status API.

Response

The response is a JSON document in the following Format. The highlighted are CDC metrics.

[{
   "requestId": "string",
   "owner": "string",
   "name": "string",
   "started": timestamp_in_milliseconds_since_epoch,
   "duration": number,
   "code": "the formatted CDC metrics",
   "type": "CDC",
   "tenant": "cdc_prod",
   "language": null,
   "messagesPerSecond": messages_per_second, 
   "latency": number,
   "maxLatency": number,
   "minLatency": number,
   "avgLatency": number,
   "maxLatencyDate": timestamp_in_milliseconds_since_epoch,
   "recordsProcessed": number,
   "recordsProcessedSinceLastCheck": number,
   "lastCheck": timestamp_in_milliseconds_since_epoch,
   "lastTimeRecordsReceivedDate": timestamp_in_milliseconds_since_epoch}
]

Response codes

200 for success, 401 for not authorized, 403 for forbidden, and 500 for an internal error.

GET CDC metrics for all running Flows

This API returns metrics for all currently running CDC Flows.

Authentication

Before making a call to API, the user must receive a JWT token from the authentication API.

Step 1. Use any user with the Administrator role to call an Etlworks Integrator authentication endpoint and receive an access token.

The API endpoint parameters:

PATH: /etl/rest/v1/tasks/?type=CDC%20stream

EXAMPLE: https://app.etlworks.com/etl/rest/v1/tasks/?type=CDC%20stream

METHOD: GET
HEADER: Authorization:Bearer access-token
REQUEST BODY: none
REQUEST CONTENT TYPE: none

Response

The response is a JSON document in the following Format. The highlighted are CDC metrics.

[{
   "requestId": "string",
   "owner": "string",
   "name": "string",
   "started": timestamp_in_milliseconds_since_epoch,
   "duration": number,
   "code": "the formatted CDC metrics",
   "type": "CDC",
   "tenant": "cdc_prod",
   "language": null,
   "messagesPerSecond": messages_per_second, 
   "latency": number,
   "maxLatency": number,
   "minLatency": number,
   "avgLatency": number,
   "maxLatencyDate": timestamp_in_milliseconds_since_epoch,
   "recordsProcessed": number,
   "recordsProcessedSinceLastCheck": number,
   "lastCheck": timestamp_in_milliseconds_since_epoch,
   "lastTimeRecordsReceivedDate": timestamp_in_milliseconds_since_epoch}
]

Response codes

200 for success, 401 for not authorized, 403 for forbidden, and 500 for an internal error.

Articles in this section

Overview

Supported CDC topologies

Cloud, with optional SSH tunnel

Hybrid-cloud with data integration agents

On-premise

Enable change data capture for the source database

CDC connectors

CDC pipeline

Built-in CDC pipelines

CDC Extract

Stream CDC events, create files in the local or cloud storage

Create CDC connection and Flow

Add source-to-destination transformation

Optionally set destination connection, override format, and CDC Key

Schedule CDC flow

Stream CDC events to a message queue

Step-by-step process

Insert events from all tables into a single topic

Include attributes available in the original CDC event emitted by Debezium

Generate partition key for Kafka and Events Hub

SQL Server tokens

MySQL tokens

Oracle tokens

Postgres tokens

DB2 tokens

Stream CDC events from the message queue to any destination

Database-specific use cases

MySQL use cases

Legacy and Modern implementations

MySQL Snapshot modes

Add new tables to the pipeline (MySQL)

Capture data from compressed binlog

Work with spatial (geometry) data types (MySQL)

Stream CDC events from the specific binlog file, position, and optionally GTID

Enable cursor-based result set when performing an initial snapshot

Postgres use cases

Postgres Snapshot modes

Add new tables to the pipeline (Postgres)

Avoid abnormal consumption of WAL database disk space

Propagate default values

SQL Server use cases

SQL Server Snapshot modes

Add new tables to the pipeline (SQL Server)

Work with binary and spatial (geometry) data types (SQL Server)

Oracle use cases

Connect to read-only Oracle database

Use JDBC URL to configure CDC connection

Oracle Snapshot modes

Improve the performance of the snapshot

Add new tables to the pipeline (Oracle)

DB2 use cases

DB2 Server Snapshot modes

Add new tables to the pipeline (DB2)

MongoDB use cases

Use Cluster URL to browse data

Connect to sharded MongoDB cluster

Add new collections to the pipeline

Enable support for MongoDB Change Streams

Serialize CDC events as JSON documents

Ad-hoc and incremental snapshots

Ad-hoc snapshots

Incremental snapshot

Trigger incremental snapshot of the new tables

Trigger ad-hoc snapshot of the existing tables

How ad-hoc snapshot works

The format of the signal

Configure ad-hoc snapshot

Trigger add-hoc snapshot

Tips and tricks

Configure data exchange formats

Convert NULL values in the source database into empty ('') strings in CSV

Do not enclose null values in double quotes

Convert boolean true/false into numeric 1/0

Preserve end-of-line (EOL) characters in column's value

Create CSV and JSON files using the specified character encoding

Handle NULL and empty ('') values when loading CSV files into databases

NULL is the same as empty ('')

NULL is not the same as empty ('')

Working with JSON columns in the source database