Change Data Capture (CDC) from transaction log

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.

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

Read about other change replication techniques available in Etlworks Integrator.

Enable change data capture for the source database

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

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

CDC when events are sent to the message queue

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

The separate flow polls the events from the topics in the message queue, then transforms and loads data into the destination.

When to use this type of CDC

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 a requirement to have a message queue, which is a separate element of the infrastructure that is not managed by Etlworks. 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.

Extract data into the message queue

This flow extracts CDC events from the CDC-enabled database and stores them in the message queue's topics.

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 whitelisted 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 1. Create a CDC connection for the source database. Configure the following parameters:

• Include Databases(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 poll data from tens or hundreds of tables, consider using regular expressions or {tokenizing} the Whitelisted Table(s) or FROM.
• CDC Key - the CDC key is used as a topic name in a message queue to load data into. For example, if CDC key is configured as [db]_[table]_cdc_stream the flow will create a topic for each database.table and add a suffix _cdc_stream to the ti=opic name. It will stream the CDC events to that topic.
• Maximum Queue Size and Maximum Batch Size - for real production setup consider increasing both by x10.
• Enable Always generate INSERT -  when this option is enabled, the flow will add two extra columns to the stream:
  • debezium_cdc_op VARCHAR - actual CRUD operation (c- create, u- update, d - delete)
  • debezium_cdc_timestamp LONG - timestamp of the event in milliseconds 

Step 2. Create a Kafka or Azure Event Hub connection.

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

Step 4. Start creating a flow by selecting the Database to queue using CDC 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 whitelisted tables from the CDC connection, or
  • a regular expression to match the table names, or 
  • a comma-separated list of tables to poll data from, or
  • {token}
• Destination connection - Kafka or Azure Event Hub connection created in step 2
• 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.invemtory,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. Configure mapping and other transformations if needed.

Step 7. Schedule the extract flow. We recommend using a continuous run schedule type. The idea is that the extract flow runs until there are no CDC events to poll, stops for the configured number of seconds, then starts again.

Load data from the message queue when the target is a relational database

This flow reads CDC events from the message queue's topics, transforms data and loads it into the destination.

Step 1. Create a Kafka or Azure Event Hub connection. You can reuse the same connection that you created for the extract flow.

You must enter a unique name in the Group ID field.

Also, consider changing the default values for the following parameters:

• Number of retries before stop polling - The number of retries before stop polling if poll returns no records. By increasing this number, you can give the flow more time to read the events from the queue.
• Max number of records to poll - The maximum number of records to poll in one call. By changing this number, you can tune the flow for the maximum performance or minimum RAM consumption.
• Max number of records to read - The total maximum number of records to read from the queue. Set it to a reasonable number to allow the system to process records in micro-batches. If nothing is entered, the system will read records from the queue until there are no more records. By changing this number you are either increasing or decreasing the size of the micro-batches.

Step 2. Create a connection to the destination database.

Step 3. Start creating a flow by typing in queue to.

Step 4. Select Queue to the database (assuming that your destination is a database).

Step 5.  Add as many source-to-destination transformations as there are queue's topics to poll data from. Most likely you are going to have one topic per source table.

For each transformation set the following parameters:

• Source connection - queue connection created in step 1.
• Source format - a format used in the extract flow, most likely JSON.
• FROM- a topic name to poll data from.
• Destination connection - a destination database connection created in step 2.
• TO -  a destination table name
• Set the Action under MAPPING/Parameters- action defines a SQL command to gerenerate. Set Action to one of the following:
  • Record - the system will automatically generate INSERT, UPDATE or DELETE SQL statement based on the value of the debezium_cdc_op field (c, u, d) populate during extract from the CDC-enabled database
  • Record with MERGE - same as Record, except the system, will generate MERGE SQL statement for inserts and updates. Note that not all databases support native MERGE, also some databases, for example, MySQL and PostgreSQL, require a unique index for the fields that are used to uniquely identify the record.
  • Record with IfExist - same as Record with MERGE, except it uses a SELECT statement to check if record exists and then conditionally generates either INSERT or UPDATE. 
• Set the Lookup fields under MAPPING/Parameters - a comma-separated list of fields that uniquely identify the record.
  • Alternatively, you can enable the Predict Lookup Fields which, if enabled, will force the flow to use various algorithms to automatically predict the fields that uniquely identify the record. Note that it is not always possible to correctly detect the unique fields.
  • When enabling the Predict Lookup Fields (which is not always accurate) is not an option you can specify the list of table=fields pairs in the Lookup Field. Use the fully-qualified table names and ';' as a separator between table=field pairs.

    Example:

    test1.inventory=inventory_id,database_name;
    test1.payment=payment_id,database_name;
    test1.rental=rental_id,database_name;

    

Step 6. Schedule load flow to run in parallel with the extract flow. Just like with extract flow we recommend using a continuous run schedule type. 

Load data from the message queue when the target is Snowflake

This flow reads CDC events from the message queue's topics, transforms data and loads it into the Snowflake.

Step 1. Create a Kafka or Azure Event Hub connection. You can reuse the same connection that you created for the extract flow.

You must enter a unique name in the Group ID field.

Also, consider changing the default values for the following parameters:

• Number of retries before stop polling - The number of retries before stop polling if poll returns no records. By increasing this number, you can give the flow more time to read the events from the queue.
• Max number of records to poll - The maximum number of records to poll in one call. By changing this number, you can tune the flow for the maximum performance or minimum RAM consumption.
• Max number of records to read - The total maximum number of records to read from the queue. Set it to a reasonable number to allow the system to process records in micro-batches. If nothing is entered, the system will read records from the queue until there are no more records. By changing this number, you are either increasing or decreasing the size of the micro-batches.

Step 2. Create all required connections and formats.

Step 3. Start creating a flow by typing in queue to snowflake.

Step 4. Select Queue to the Snowflake.

Follow the same steps as suggested in this article.

Step 5.  Add as many source-to-destination transformations as there are queue's topics to poll data from. Most likely you are going to have one topic per source table.

For each transformation set the following parameters:

• Source connection - queue connection created in step 1.
• Source format - a format used in the extract flow, most likely JSON.
• FROM- a topic name to poll data from.
• Destination connection - a connection for the Snowflake stage.
• Destination format - CSV format 
• TO -  a destination table name

Step 6. All parameters are the same as in the regular Snowflake flow, except the Action, which can be set to the one of the following: 

• CDC MERGE - the system will use the event type stored in the record: "c" for INSERT, "u" for UPDATE and "d" for DELETE. This option requires setting the Lookup fields to the comma-separated list of fields that uniquely identify the record in the target database. Optionally you can enable the Predict Lookup Fields option. If neither works you can specify the list of table=fields pairs in the Lookup Field. Use the fully-qualified table names and ';' as a separator between table=field pairs.
• COPY INTO - in some cases, it makes sense to always insert unmodified CDC events into the staging table. Then load events from the staging table(s) into the final table using SQL. When this option is enabled, the following fields will be automatically added to the staging table:
  • debezium_cdc_op - event type, "c" for create, "u" for update and "d" for delete.
  • debezium_cdc_timestamp - the event timestamp in milliseconds. 

Step 7. Schedule load flow to run in parallel with the extract flow. Just like with extract flow we recommend using a continuous run schedule type. 

Load data from the message queue when the target is Amazon Redshift

This flow reads CDC events from the message queue's topics, transforms data and loads it into the Amazon Redshift.

Step 1. Create a Kafka or Azure Event Hub connection. You can reuse the same connection that you created for the extract flow.

You must enter a unique name in the Group ID field.

Also, consider changing the default values for the following parameters:

• Number of retries before stop polling - The number of retries before stop polling if poll returns no records. By increasing this number, you can give flow more time to read the events from the queue.
• Max number of records to poll - The maximum number of records to poll in one call. By changing this number, you can tune the flow for the maximum performance or minimum RAM consumption.
• Max number of records to read - The total maximum number of records to read from the queue. Set it to a reasonable number to allow the system to process records in micro-batches. If nothing is entered, the system will read records from the queue until there are no more records. By changing this number you are either increasing or decreasing the size of the micro-batches.

Step 2. Create all required connections and formats.

Step 3. Start creating a flow by typing in queue to redshift.

Step 4. Select Queue to the Redshift.

Follow the same steps as suggested in this article.

Step 5.  Add as many source-to-destination transformations as there are queue's topics to poll data from. Most likely you are going to have one topic per source table.

For each transformation set the following parameters:

• Source connection - queue connection created in step 1.
• Source format - a format used in the extract flow, most likely JSON.
• FROM- a topic name to poll data from.
• Destination connection - a connection for the Redshift stage.
• Destination format - CSV format 
• TO -  a destination table name

Step 6. All parameters are the same as in the regular Redshift flow, except the Action, which can be set to the one of the following: 

• CDC MERGE - the system will use the event type stored in the record: "c" for INSERT, "u" for UPDATE and "d" for DELETE. This option requires setting the Lookup fields to the comma-separated list of fields that uniquely identify the record in the target database. Optionally, you can enable the Predict Lookup Fields option. If neither works you can specify the list of table=fields pairs in the Lookup Field. Use the fully-qualified table names and ';' as a separator between table=field pairs.
• COPY INTO - in some cases, it makes sense to always insert unmodified CDC events into the staging table. Then load events from the staging table(s) into the final table using SQL. When this option is enabled, the following fields will be automatically added to the staging table:
  • debezium_cdc_op - event type, "c" for create, "u" for update and "d" for delete.
  • debezium_cdc_timestamp - the event timestamp in milliseconds. 

Step 7. Schedule load flow to run in parallel with the extract flow. Just like with extract flow, we recommend using a continuous run schedule type. 

CDC when events are stored as CSV files

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

The separate flow reads the events stored in CSV files, then transforms and loads data into the destination. The load flow must delete the processed files.

When to use this type of CDC

This flow type is very similar to the CDC with a message queue but does not require any additional elements of the infrastructure. It is also the fastest type of CDC supported by Etlworks.

Extract data into the CSV files in local storage

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 whitelisted 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 1. Create a CDC connection for the source database. Specifically, configure the following parameters:

• Enable Serialize CDC events as CSV files into internal storage 
• Include Databases(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 poll data from tens or hundreds of tables consider using regular expressions or {tokenizing} the Whitelisted Table(s) or FROM.
• CDC Key - the CDC key will be used as the name of the CSV file that contains the CDC events. For example, if CDC key is configured as [db]_[table]_cdc_stream the flow will create a file for each database.table with a suffix _cdc_stream and will stream the CDC events to this file.
• Enable Treat 'null' as NULL
• Maximum Queue Size and Maximum Batch Size - for real production setup consider increasing both by at least x10.
• Enable Always generate INSERT -  when this option is enabled the flow will add two extra columns to the stream:
  • debezium_cdc_op VARCHAR - actual CRUD operation (c- create, u- update, d - delete)
  • debezium_cdc_timestamp LONG - timestamp of the event in milliseconds 

Step 2.  Create new flow using flow type Stream CDC events

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

• Source connection - CDC connection created in step 1
• FROM - the following options are available:
  • * - the system will get the list of the whitelisted tables from the CDC connection, or
  • a regular expression to match the table names, or 
  • a comma-separated list of tables to poll data from, or
  • {token}

Example with a comma-separated list of whitelisted 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 files. You can override it by changing the value of the property Location of the CDC events serialized as CSV files:

Step 4. Schedule the extract flow. We recommend using a continuous run schedule type. The idea is that the extract flow runs until there are no CDC events to poll, stops for the configured number of seconds, then starts again.

Load data from CSV files when the target is a relational database

This flow reads CDC events from the CSV files in local storage, transforms data and loads it into the destination. 

The staged CSV files created by the extract flow are moved to the ready folder, processed from there, then deleted after the processing.

Step 1. Create a Server storage connection for the staged CSV files. The default location is {app.data}/debezium_data/events.

Step 2. Create a Server storage connection for the ready CSV files. The default location is {app.data}/debezium_data/events/ready.

Step 3. Create CSV format with all default settings.

Step 4. Create a connection to the destination database.

Step 5. Create Move files flow where the source is a connection created in step 1 and the destination is a connection created in step 2. Use *_cdc_stream_*.csv as wildcard file name for the source (FROM):

Optionally limit the number of files to move in one batch by setting the number of files in MAPPING->Parameters->Maximum Number of Files to Process. Limiting the number of files will allow the system to handle a large number of files that can be potentially created by the extract flow without overflowing the files queue.

Step 6. Create a new flow using flow type file to database (assuming that the target is a database).

Step 7.  Add as many source-to-destination transformations as there are distinct filename patterns to poll data from.

For example, if the following files are created by the extract flow:

• {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

you will need 3 pairs of source-to-destination transformations, where the sources (FROM) will need to be set to the following wildcard file names:

• test_inventory_cdc_stream_*.csv
• test_payment_cdc_stream_*.csv
• test_customer_cdc_stream_*.csv

Alternatively, if you want to process all files without creating separate pairs of source-to-destination transformations: 

• Set the FROM to the wildcard file name that matches all files to load, for example *_*_cdc_stream_*.csv
• Set the TO to schema.*, for example public.*
• Set Calculate Destination Object Name under MAPPING->Parameters to

var start = name.indexOf('_');
var end = name.indexOf('_cdc_stream');

value = name.substring(start + 1, end);

You can also use regular expression _(.*?)_cdc_stream to match the table names.

Read more about configuring TO when processing files by a wildcard name.

Step 8. For each transformation set the following parameters:

• Source connection - ready connection created in step 2.
• CSV format created in step 3.
• FROM- a wildcard filename as described above.
• Destination connection - a destination database connection created in step 4.
• TO -  a destination table name or a wildcard table name as described above
• Enable Ignore when there is no file and Process all files under the MAPPING/Parameters
• Enable Delete loaded sources files
• Set the Action under MAPPING/Parameters- action defines a SQL command to generate. Set Action to one of the following: 
  • Record - the system will automatically generate INSERT, UPDATE or DELETE SQL statement based on the value of the debezium_cdc_op field (c, u, d) populate during extract from the CDC-enabled database
  • Record with MERGE - same as Record, except the system, will generate MERGE SQL statement for inserts and updates. Note that not all databases support native MERGE, also some databases, for example, MySQL and PostgreSQL, require a unique index for the fields that are used to uniquely identify the record.
  • Record with IfExist - same as Record with MERGE, except it uses a SELECT statement to check if record exists and then conditionally generates either INSERT or UPDATE. 
• Set the Lookup fields under MAPPING/Parameters - a comma-separated list of fields that uniquely identify the record. Alternatively, you can enable the Predict Lookup Fields which if enabled will force the flow to use various algorithms to automatically predict the fields that uniquely identify the record. Note that it is not always possible to correctly detect the unique fields. If neither works you can specify the list of table=fields pairs in the Lookup Field. Use the fully-qualified table names and ';' as a separator between table=field pairs.

Step 9. Create nested flow and add flows created in the previous steps in the following order:

1. Move to ready
2. Load

Step 10. Schedule load flow to run in parallel with the extract flow. Just like with extract flow we recommend using a continuous run schedule type. 

Load data from CSV files when the target is Snowflake

This flow reads CDC events from the CSV files in local storage, transforms data and loads it into the Snowflake. 

The staged CSV files created by the extract flow are moved to the ready folder, processed from there, then deleted after the processing.

Step 1. Create a Server storage connection for the staged CSV files. The default location is {app.data}/debezium_data/events.

Step 2. Create a Server storage connection for the ready CSV files. The default location is {app.data}/debezium_data/events/ready.

Step 3. Create CSV format with all default settings.

Step 4. Create all required for Snowflake flows connections and formats.

Step 5. Create Move files flow where the source is a connection created in step 1 and the destination is a connection created in step 2. Use *_cdc_stream_*.csv as wildcard file name for the source (FROM):

Optionally limit the number of files to move in one batch by setting the number of files in MAPPING->Parameters->Maximum Number of Files to Process. Limiting the number of files will allow the system to handle a large number of files that can be potentially created by the extract flow without overflowing the files queue.

Step 6. Create a new flow using flow type file to snowflake 

Follow the same steps as suggested in this article.

Step 7.  Add as many source-to-destination transformations as there are distinct filename patterns to poll data from.

For example, if the following files are created by the extract flow:

• {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

you will need 3 pairs of source-to-destination transformations, where the sources (FROM) will need to be set to the following wildcard file names:

• test_inventory_cdc_stream_*.csv
• test_payment_cdc_stream_*.csv
• test_customer_cdc_stream_*.csv

Alternatively, if you want to process all files without creating separate pairs of source-to-destination transformations: 

• Set the FROM to the wildcard file name that matches all files to load, for example *_*_cdc_stream_*.csv
• Set the TO to schema.*, for example public.*
• Set Calculate Destination Object Name under MAPPING->Parameters to

var start = name.indexOf('_');
var end = name.indexOf('_cdc_stream');

value = name.substring(start + 1, end);

You can also use regular expression _(.*?)_cdc_stream to match the table names.

Read more about configuring TO when processing files by a wildcard name.

For each transformation set the following parameters:

• Source connection - ready connection created in step 2.
• CSV format created in step 3.
• FROM- a wildcard filename as described above.
• Destination connection - a connection for the Snowflake stage.
• Destination format - CSV format 
• TO -  a destination table name

Step 8. Enable Process all files under Parameters

Step 9. If you are not planning to transform the data before loading into the Snowflake enable Ignore Transformations under Parameters. This will have a positive impact on performance.

Step 10. Enable Delete loaded source files under Parameters.

Step 11. All other parameters are the same as in the regular Snowflake flow, except the Action, which can be set to the one of the following: 

• CDC MERGE - the system will use the event type stored in the record: "c" for INSERT, "u" for UPDATE and "d" for DELETE. This option requires setting the Lookup fields to the comma-separated list of fields that uniquely identify the record in the target database. Optionally, you can enable the Predict Lookup Fields option. If neither works you can specify the list of table=fields pairs in the Lookup Field. Use the fully-qualified table names and ';' as a separator between table=field pairs.
• COPY INTO - in some cases, it makes sense to always insert unmodified CDC events into the staging table. Then load events from the staging table(s) into the final table using SQL. When this option is enabled, the following fields will be automatically added to the staging table:
  • debezium_cdc_op - event type, "c" for create, "u" for update and "d" for delete.
  • debezium_cdc_timestamp - the event timestamp in milliseconds. 

Step 12. Create nested flow and add flows created in the previous steps in the following order:

1. Move to ready
2. Load

Step 13. Schedule load flow to run in parallel with the extract flow. Just like with extract flow we recommend using a continuous run schedule type. 

Load data from CSV files when the target is Amazon Redshift

This flow reads CDC events from the CSV files in local storage, transforms data and loads it into the Amazon Redshift. 

The staged CSV files created by the extract flow are moved to the ready folder, processed from there, then deleted after the processing.

Step 1. Create a Server storage connection for the staged CSV files. The default location is {app.data}/debezium_data/events.

Step 2. Create a Server storage connection for the ready CSV files. The default location is {app.data}/debezium_data/events/ready.

Step 3. Create CSV format with all default settings.

Step 4. Create all required for Redshift flows connections and formats.

Step 5. Create Move files flow where the source is a connection created in step 1 and the destination is a connection created in step 2. Use *_cdc_stream_*.csv as wildcard file name for the source (FROM):

Optionally limit the number of files to move in one batch by setting the number of files in MAPPING->Parameters->Maximum Number of Files to Process. Limiting the number of files will allow the system to handle a large number of files that can be potentially created by the extract flow without overflowing the files queue.

Step 6. Create a new flow using flow type file to redshift

Follow the same steps as suggested in this article.

Step 7.  Add as many source-to-destination transformations as there are distinct filename patterns to poll data from.

For example, if the following files are created by the extract flow:

• {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

you will need 3 pairs of source-to-destination transformations, where the sources (FROM) will need to be set to the following wildcard file names:

• test_inventory_cdc_stream_*.csv
• test_payment_cdc_stream_*.csv
• test_customer_cdc_stream_*.csv

Alternatively, if you want to process all files without creating separate pairs of source-to-destination transformations: 

• Set the FROM to the wildcard file name that matches all files to load, for example *_*_cdc_stream_*.csv
• Set the TO to schema.*, for example public.*
• Set Calculate Destination Object Name under MAPPING->Parameters to

var start = name.indexOf('_');
var end = name.indexOf('_cdc_stream');

value = name.substring(start + 1, end);

You can also use regular expression _(.*?)_cdc_stream to match the table names.

Read more about configuring TO when processing files by a wildcard name.

For each transformation set the following parameters:

• Source connection - ready connection created in step 2.
• CSV format created in step 3.
• FROM- a wildcard filename as described above.
• Destination connection - a connection for the Snowflake stage.
• Destination format - CSV format 
• TO -  a destination table name

Step 8. Enable Process all files under Parameters

Step 9. If you are not planning to transform the data before loading into the Redshift enable Ignore Transformations under Parameters. This will have a positive impact on performance.

Step 10. Enable Delete loaded source files under Parameters.

Step 11. All other parameters are the same as in the regular Redshift flow, except the Action, which can be set to the one of the following: 

• CDC MERGE - the system will use the event type stored in the record: "c" for INSERT, "u" for UPDATE and "d" for DELETE. This option requires setting the Lookup fields to the comma-separated list of fields that uniquely identify the record in the target database. Optionally, you can enable the Predict Lookup Fields option. If neither works you can specify the list of table=fields pairs in the Lookup Field. Use the fully-qualified table names and ';' as a separator between table=field pairs.
• COPY - in some cases, it makes sense to always insert unmodified CDC events into the staging table. Then load events from the staging table(s) into the final table using SQL. When this option is enabled the following fields will be automatically added to the staging table:
  • debezium_cdc_op - event type, "c" for create, "u" for update and "d" for delete.
  • debezium_cdc_timestamp - the event timestamp in milliseconds. 

 

Step 12. Create nested flow and add flows created in the previous steps in the following order:

1. Move to ready
2. Load

Step 13. Schedule load flow to run in parallel with the extract flow. Just like with extract flow we recommend using a continuous run schedule type. 

Point-to-point CDC

The point-to-point CDC connects the CDC-enabled source directly to the destination. 

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

When to use this type of CDC

There is no message queue or CSV files in a middle, also no separate extract and load flows. At the same time, the CDC stream typically contains a mix of events from different sources (for example updates to the different tables) so this flow type is only useful when you are willing to separate each source-destination pair into its own independently scheduled flow or make every source-to-destination transformation in a flow run in a parallel.

We recommend using this flow type for quick prototyping or you when you can afford to have a separate flow for each source-to-destination pair.

Point-to-point CDC when the target is a relational database

Step 1. Create a CDC connection for the source database.

Step 2. Create a destination connection. 

Step 3. Create a new flow by selecting any of the following CDC flows from the gallery.

Step 4. Add a new source-to-destination transformation where:

• the FROM connection is the CDC connection created in step 1.
• the FROM is a fully qualified source table name (test.dbo.patient). When the source is a MongoDB connection enter or select the MongoDB collection name. 
• the TO is a destination table name. 
• the TO connection is the destination database connection created in step 2.

Step 5. Click the MAPPING button and select the Parameters tab. Set Action to one of the following options:

• Record - the system will use the event type stored in the record: "c" for INSERT, "u" for UPDATE and "d" for DELETE. This option requires setting the Lookup fields to the comma-separated list of fields that uniquely identify the record in the target database. Optionally you can enable the Predict Lookup Fields option.
• Record with MERGE - works the same as Record except always executing the native MERGE (UPSERT) on the target database for the "c" and "u" events.
• Record with IfExists - works the same as Record except always checking if the record exists in the target database and then conditionally excusing INSERT or UPDATE. Use this option if the native MERGE is not supported by the target database.
• INSERT - in some cases, it makes sense to always insert unmodified CDC events into some sort of staging table. Then load events from the staging table(s) into the final table using SQL. When this option is enabled the following fields will be automatically added to the staging table:
  • 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. 
• To enable the INSERT option:
• 1. Enable the Always generate INSERT for the source CDC connection:
• 
• 2. Set the Action to INSERT.

Step 6. Schedule load flow to run in parallel with the extract flow. Just like with extract flow, we recommend using a continuous run schedule type. 

Point-to-point CDC when the target is Snowflake

Step 1. Create a source CDC connection for the source database.

enable the Always generate INSERT.

Step 2. Create a connection to the Snowflake. Make sure the auto-commit is enabled.

Step 3. Create a new flow by selecting any of the CDC flows from the gallery.

Step 4. All parameters are the same as in the regular Snowflake flow, except the Action, which can be set to the one of the following: 

• CDC MERGE - the system will use the event type stored in the record: "c" for INSERT, "u" for UPDATE and "d" for DELETE. This option requires setting the Lookup fields to the comma-separated list of fields that uniquely identify the record in the target database. Optionally you can enable the Predict Lookup Fields option.
• COPY INTO - in some cases, it makes sense to always insert unmodified CDC events into the staging table. Then load events from the staging table(s) into the final table using SQL. When this option is enabled the following fields will be automatically added to the staging table:
  • debezium_cdc_op - event type, "c" for create, "u" for update and "d" for delete.
  • debezium_cdc_timestamp - the event timestamp in milliseconds. 

Point-to-point CDC when the target is Amazon Redshift

Step 1. Create a source CDC connection for the source database.

Enable the Always generate INSERT.

Step 2. Create a connection to the Redshift. Make sure the auto-commit is enabled.

Step 3. Create a new flow by selecting any of the CDC flows from the gallery.

 

Step 4. All parameters are the same as in the regular Redshift flow, except the Action, which can be set to the one of the following: 

• CDC MERGE - the system will use the event type stored in the record: "c" for INSERT, "u" for UPDATE and "d" for DELETE. This option requires setting the Lookup fields to the comma-separated list of fields that uniquely identify the record in the target database. Optionally you can enable the Predict Lookup Fields option.
• COPY - in some cases, it makes sense to always insert unmodified CDC events into some sort of staging table. Then load events from the staging table(s) into the final table using SQL. When this option is enabled, the following fields will be automatically added to the staging table:
  • debezium_cdc_op - event type, "c" for create, "u" for update and "d" for delete.
  • debezium_cdc_timestamp - the event timestamp in milliseconds. 

Handling Errors

These options are available for all flavors of CDC flows but really make sense only for point-to-point CDC.

Restream from the beginning if Error

Typically, the CDC flow will perform a full load in the first run and will automatically switch to the CDC mode for all consecutive runs. 

You can enable option Restream from the beginning if Error when configuring the CDC connection. 

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. Use it with caution. 

Use Internal Queue for better reliability 

When configuring the CDC connection, it is possible to enable the internal queue which will be used to temporarily store all CDC events until they are successfully loaded into the target.

To enable or disable the queue, open the CDC connection and select one of the available options for the field Use Internal Queue:

• never - disable the queue. 
• always - enable the queue for snapshot and incremental pull.
• snapshot only - enable the queue for the snapshot only.
• except snapshot - enable the queue for the incremental pull only.

This option is ignored if Restream from the beginning if Error is enabled.

The sequence diagram below demonstrates how CDC events are getting processed when the internal queue is enabled.