Connectors for Change Data Capture (CDC)

When to use change data capture connectors

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).

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

The CDC connection is used as a source in the:

• CDC flows
• Snowflake CDC flows
• Redshift CDC flows

Creating a connection

Step 1. Enable CDC for the source database.

• Microsoft SQL Server
• MySQL
• Oracle
• PostgreSQL
• MongoDB

Step 2. In the Connections window, click the + button and type in cdc.

Step 3. Select the source database.

Step 3. Enter connection parameters.

Connection parameters

All databases 

• Snapshot Mode - it is different for each database (see below).
• Restream from the beginning if Error - if an error occurred and this option is enabled the system will automatically switch to the snapshot mode and attempt to reload the whole table in the next run. Read more about handling errors in CDC flows. 
• Use Internal Queue - If the use of queue is enabled the system will store CDC events in the queue before sending them further in the pipeline for processing. The queue is used to reprocessed events in case of failures. This option is ignored if Restream from the beginning if Error is enabled. Read more about using internal queue to improve the reliability of the CDC flows.
• Always generate INSERT - if enabled, the system will always generate INSERT SQL, even for UPDATEs and DELETEs. When this option is enabled the following fields will be automatically added to the downstream payload:
  • debezium_cdc_op - event type, "c" for create, "u" for update and "d" for delete.
  • debezium_cdc_timestamp - the event timestamp encoded as EPOCH time in milliseconds. 
• This field must be enabled when loading data into an online data warehouses, such as Amazon Redshift or Snowflake.
• Extra columns - the columns specified in this field will be added to the end of the stream. The column value will be either empty or set to the value of the global variable with the same name.
• Number of retries before giving up - the number of retries before giving up if poll returns no records. The default is 0
• Flash Interval (ms) - this parameter defines how often (in milliseconds) the system flashes the offset pointer. The offset pointer is used to thack the last known position of the CDC records in the database transaction log. 
• Maximum Queue Size - a positive integer value that specifies the maximum size of the blocking queue into which change events read from the database log are placed before they are written to stream. This queue can provide backpressure to the transactional log reader when, for example, writes to the stream are slower. Events that appear in the queue are not included in the offsets periodically recorded by this connector. Defaults to 8192, and should always be larger than the maximum batch size.
• Maximum Batch Size - positive integer value that specifies the maximum size of each batch of events that should be processed during each iteration of this connector. Defaults to 2048.
• Poll Interval in Milliseconds - positive integer value that specifies the number of milliseconds the connector should wait during each iteration for new change events to appear. Defaults to 1000 milliseconds, or 1 second.
• Connection Timeout in Milliseconds - a positive integer value that specifies the maximum time in milliseconds this connector should wait after trying to connect to the database server before timing out. Defaults to 30 seconds.
• Other Parameters - other parameters as key=value pairs. The parameters are specific for the different connectors.
• Date and Time Format - the format for TIMESTAMP fields. If not entered the TIMESTAMP fields will be serialized as Linux epoch time (in milliseconds).
• Date Format - the format for DATE fields. If not entered the DATE fields will be serialized as Linux epoch time (in milliseconds).
• Time Format - the format for TIME fields. If not entered the TIME fields will be serialized as Linux epoch time (in milliseconds).

Connecting to a database over SSH tunnel

If you need to access a database that can only be accessed via an SSH tunnel, you need to specify additional information in the "Connect over SSH tunnel" section of the database connection screen.

• SSH Host - the name or IP address for the host accepting SSH connections.
• SSH Port - the port accepting SSH connections. The default value is 22.
• SSH User  - the user name
• SSH Password - the optional password
• Private Key File - the private key file in the pem or ppk format used for SSH authentication. Click the "Upload/Select ssh key" button to manage SSH keys using GUI. You can also upload the private key file manually and use the token {app.data} as a part of the filename, for example, {app.data}/keys/secret.pem. This parameter is optional.

SQL Server

• Host - the host or IP address of the source database
• Port - the port the source database is listening to
• User - the user name
• Password - the password
• Database - the name of the database from which to stream the changes.
• Table - the fully qualified name of the source table. This parameter will be overridden in the TO field of the CDC flow.
• JDBC Fetch Size - the JDBC Fetch Size field can be used to set the maximum number of records which will be retrieved in one database call when fetching the data in a snapshot mode, thus limiting the memory consumption within the JVM. The default is 10000.
• Max Rows to Read - the maximum number of rows to read when fetching the data in a snapshot mode. The default is empty, hence no limit.
• Max Field Size - sets the limit for the maximum number of bytes that can be returned for character and binary column values in a ResultSet object produced by this Statement object. The default is empty, hence no limit.
• Server - logical name that identifies and provides a namespace for the particular SQL Server database server being monitored. The logical name should be unique across all other connectors.
• Snapshot Mode - a mode for taking an initial snapshot of the structure and optional data of captured tables. Supported values are initial (will take a snapshot of structure and data of captured tables; useful if topics should be populated with a complete representation of the data from the captured tables) and initial_schema_only (will take a snapshot of the structure of captured tables only; useful if only changes happening from now onwards should be propagated to topics). Once the snapshot is complete, the connector will continue reading change events from the database’s redo logs.

MySQL

• Host - the host or IP address of the source database
• Port - the port the source database is listening to
• User - the user name
• Password - the password
• Database - the name of the database from which to stream the changes.
• Table - the fully qualified name of the source table. This parameter will be overridden in the TO field of the CDC flow.
• JDBC Fetch Size - the JDBC Fetch Size field can be used to set the maximum number of records which will be retrieved in one database call when fetching the data in a snapshot mode, thus limiting the memory consumption within the JVM. The default is 10000.
• Max Rows to Read - the maximum number of rows to read when fetching the data in a snapshot mode. The default is empty, hence no limit.
• Max Field Size - sets the limit for the maximum number of bytes that can be returned for character and binary column values in a ResultSet object produced by this Statement object. The default is empty, hence no limit.
• Server - logical name that identifies and provides a namespace for the particular MySQL database server/cluster being monitored. The logical name should be unique across all other connectors. Defaults to host:port, though we recommend using an explicit and meaningful logical name.
• Case sensitive matching - By default, the system will use the case insensitive pattern to whitelist the db.table to stream. For example, if there is a database 'test' and the tables 'abc' and 'Abc' the stream will include records from 'test.abc' and 'test.Abc'. By enabling this option you can enforce the case sensitive matching of the db.table defined for the source.
• Snapshot Mode - specifies the criteria for running a snapshot upon startup of the connector. The default is initial, and specifies the connector can run a snapshot only when no offsets have been recorded for the logical server name. The when_needed option specifies that the connector run a snapshot upon startup whenever it deems it necessary (when no offsets are available, or when a previously recorded offset specifies a binlog location or GTID that is not available in the server). The never option specifies that the connect should never use snapshots and that upon first startup with a logical server name the connector should read from the beginning of the binlog. This should be used with care, as it is only valid when the binlog is guaranteed to contain the entire history of the database. If you don’t need the topics to contain a consistent snapshot of the data but only need them to have the changes since the connector was started, you can use the schema_only option, where the connector only snapshots the schemas (not the data). schema_only_recovery is a recovery option for an existing connector to recover a corrupted or lost database history topic, or to periodically clean up a database history topic (which requires infinite retention) that may be growing unexpectedly.

Oracle

• Host - the host or IP address of the source database
• Port - the port the source database is listening to
• User - the user name
• Password - the password
• Database - the name of the database from which to stream the changes.
• Table - the fully qualified name of the source table. This parameter will be overridden in the TO field of the CDC flow.
• JDBC Fetch Size - the JDBC Fetch Size field can be used to set the maximum number of records which will be retrieved in one database call when fetching the data in a snapshot mode, thus limiting the memory consumption within the JVM. The default is 10000.
• Max Rows to Read - the maximum number of rows to read when fetching the data in a snapshot mode. The default is empty, hence no limit.
• Max Field Size - sets the limit for the maximum number of bytes that can be returned for character and binary column values in a ResultSet object produced by this Statement object. The default is empty, hence no limit.
• Server - logical name that identifies and provides a namespace for the particular Oracle database server being monitored. The logical name should be unique across all other connectors.
• Database - the name of the Oracle database from which to stream the changes. Must be the CDB name when working with the CDB + PDB model.
• Name of the PDB - name of the PDB to connect to, when working with the CDB + PDB model.
• Name of the XStream outbound server - name of the XStream outbound server configured in the database.
• Snapshot Mode - a mode for taking an initial snapshot of the structure and optional data of captured tables. Supported values are initial (will take a snapshot of structure and data of captured tables; useful if topics should be populated with a complete representation of the data from the captured tables) and initial_schema_only (will take a snapshot of the structure of captured tables only; useful if only changes happening from now onwards should be propagated to topics). Once the snapshot is complete, the connector will continue reading change events from the database’s redo logs.

Postgres

• Host - the host or IP address of the source database
• Port - the port the source database is listening to
• User - the user name
• Password - the password
• Database - the name of the database from which to stream the changes.
• Table - the fully qualified name of the source table. This parameter will be overridden in the TO field of the CDC flow.
• JDBC Fetch Size - the JDBC Fetch Size field can be used to set the maximum number of records which will be retrieved in one database call when fetching the data in a snapshot mode, thus limiting the memory consumption within the JVM. The default is 10000.
• Max Rows to Read - the maximum number of rows to read when fetching the data in a snapshot mode. The default is empty, hence no limit.
• Max Field Size - sets the limit for the maximum number of bytes that can be returned for character and binary column values in a ResultSet object produced by this Statement object. The default is empty, hence no limit.
• Server - logical name that identifies and provides a namespace for the particular PostgreSQL database server/cluster being monitored. The logical name should be unique across all other connectors, since it is used as a prefix for all Kafka topic names coming from this connector. Defaults to host:port/dbname, though we recommend using an explicit and meaningful logical name.
• Output Plugin - PostgreSQL’s logical decoding feature was first introduced in version 9.4 and is a mechanism which allows the extraction of the changes which were committed to the transaction log and the processing of these changes in a user-friendly manner via the help of an output plugin. This output plugin must be installed prior to running the PostgreSQL server and enabled together with a replication slot in order for clients to be able to consume the changes.
• Snapshot Mode - specifies the criteria for running a snapshot upon startup of the connector. The default is initial, and specifies the connector can run a snapshot only when no offsets have been recorded for the logical server name. The always option specifies that the connector run a snapshot each time on startup. The never option specifies that the connect should never use snapshots and that upon first startup with a logical server name the connector should read from either from where it last left off (last LSN position) or start from the beginning from the point of the view of the logical replication slot. Finally, the initial_only option specifies that the connector should only take an initial snapshot and then stop, without processing any subsequent changes.

MongoDB

• Cluster URL - the required connection string. Read about the MongoDB connection string.
• Replica Sets or Shards - to use the MongoDB connector with a replica set, simply provide the addresses of one or more replica set servers as seed addresses. The connector will use these seeds to connect to the replica set, and then once connected will get from the replica set the complete set of members and which member is primary. The connector will start a task to connect to the primary and capture the changes from the primary’s oplog. When the replica set elects a new primary, the task will automatically switch over to the new primary. To use the MongoDB connector with a sharded cluster, configure the connector with the host addresses of the configuration server replica set. When the connector connects to this replica set, it discovers that it is acting as the configuration server for a sharded cluster, discovers the information about each replica set used as a shard in the cluster, and will then start up a separate task to capture the changes from each replica set. If new shards are added to the cluster or existing shards removed, the connector will automatically adjust its tasks accordingly.
• User - the optional MongoDB user name.
• Password - the optional password.
• Name - a unique name that identifies the MongoDB replica set or sharded cluster that this connector monitors. Defaults to URL, though we recommend using an explicit and meaningful logical name.
• Database - the name of the MongoDB database to stream changes from.
• Collection - the MongoDB collection to stream changes from. This parameter can be overwritten in the flow.
• Snapshot Mode - specifies the criteria for running a snapshot (eg. initial sync) upon startup of the connector. The default is initial, and specifies the connector reads a snapshot when either no offset is found or if the oplog no longer contains the previous offset. The never option specifies that the connector should never use snapshots, instead the connector should proceed to tail the log.
• What to do with the Object ID (_id) when reading - using this parameter you can configure how the system will work with the MongoDB object id (_id field) when reading the document. The following options are available:
  • remove (default) - the system will remove the _id field from the dataset
  • keep - the system will keep the _id field unmodified
  • flatten - the system will flatten the value of the _id field.  Example: "_id": ObjectId("54759eb3c090d83494e2d804") to "_id": "54759eb3c090d83494e2d804"
• Number of Documents in Explorer - The maximum number of the MongoDB documents that can be displayed in the Explorer. The default value is 10, the maximum possible value is 9999. MongoDB can contain millions of documents, so displaying all of them can kill the server.