Db2 CDC Connector #

The Db2 CDC connector allows for reading snapshot data and incremental data from Db2 database. This document describes how to setup the db2 CDC connector to run SQL queries against Db2 databases.

Supported Databases #

Dependencies #

In order to set up the Db2 CDC connector, the following table provides dependency information for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.

Maven dependency #

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-db2-cdc</artifactId> <version>3.5.0</version> </dependency>

SQL Client JAR #

Download flink-sql-connector-db2-cdc-3.5.0.jar and put it under <FLINK_HOME>/lib/.

Note: Refer to flink-sql-connector-db2-cdc, more released versions will be available in the Maven central warehouse.

Since Db2 Connector’s IPLA license is incompatible with Flink CDC project, we can’t provide Db2 connector in prebuilt connector jar packages. You may need to configure the following dependencies manually.

Setup Db2 server #

Follow the steps in the Debezium Db2 Connector.

Notes #

Not support BOOLEAN type in SQL Replication on Db2 #

Only snapshots can be taken from tables with BOOLEAN type columns. Currently, SQL Replication on Db2 does not support BOOLEAN, so Debezium can not perform CDC on those tables. Consider using another type to replace BOOLEAN type.

How to create a Db2 CDC table #

The Db2 CDC table can be defined as following:

-- checkpoint every 3 seconds Flink SQL> SET 'execution.checkpointing.interval' = '3s';  -- register a Db2 table 'products' in Flink SQL Flink SQL> CREATE TABLE products (  ID INT NOT NULL,  NAME STRING,  DESCRIPTION STRING,  WEIGHT DECIMAL(10,3),  PRIMARY KEY(ID) NOT ENFORCED  ) WITH (  'connector' = 'db2-cdc',  'hostname' = 'localhost',  'port' = '50000',  'username' = 'root',  'password' = '123456',  'database-name' = 'mydb',  'table-name' = 'myschema.products');  -- read snapshot and redo logs from products table Flink SQL> SELECT * FROM products; 

Connector Options #

Available Metadata #

The following format metadata can be exposed as read-only (VIRTUAL) columns in a table definition.

Features #

Startup Reading Position #

The config option scan.startup.mode specifies the startup mode for DB2 CDC consumer. The valid enumerations are:

• initial (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest redo logs.
• latest-offset: Never to perform snapshot on the monitored database tables upon first startup, just read from the end of the redo logs which means only have the changes since the connector was started.

Note: the mechanism of scan.startup.mode option relying on Debezium’s snapshot.mode configuration. So please do not using them together. If you speicifying both scan.startup.mode and debezium.snapshot.mode options in the table DDL, it may make scan.startup.mode doesn’t work.

DataStream Source #

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.source.SourceFunction;  import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;  public class Db2SourceExample {  public static void main(String[] args) throws Exception {  SourceFunction<String> db2Source =  Db2Source.<String>builder()  .hostname("yourHostname")  .port(50000)  .database("yourDatabaseName") // set captured database  .tableList("yourSchemaName.yourTableName") // set captured table  .username("yourUsername")  .password("yourPassword")  .deserializer(  new JsonDebeziumDeserializationSchema()) // converts SourceRecord to  // JSON String  .build();   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();   // enable checkpoint  env.enableCheckpointing(3000);   env.addSource(db2Source)  .print()  .setParallelism(1); // use parallelism 1 for sink to keep message ordering   env.execute("Print Db2 Snapshot + Change Stream");  } } 

The DB2 CDC incremental connector (since 3.1.0) can be used as the following shows:

import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;  import org.apache.flink.cdc.connectors.base.options.StartupOptions; import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder; import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder.Db2IncrementalSource; import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;  public class Db2ParallelSourceExample {   public static void main(String[] args) throws Exception {   Db2IncrementalSource<String> sqlServerSource =  new Db2SourceBuilder()  .hostname("localhost")  .port(50000)  .databaseList("TESTDB")  .tableList("DB2INST1.CUSTOMERS")  .username("flink")  .password("flinkpw")  .deserializer(new JsonDebeziumDeserializationSchema())  .startupOptions(StartupOptions.initial())  .build();   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  // enable checkpoint  env.enableCheckpointing(3000);  // set the source parallelism to 2  env.fromSource(sqlServerSource, WatermarkStrategy.noWatermarks(), "Db2IncrementalSource")  .setParallelism(2)  .print()  .setParallelism(1);   env.execute("Print DB2 Snapshot + Change Stream");  } } 

Available Source metrics #

Metrics can help understand the progress of assignments, and the following are the supported Flink metrics:

Notice:

1. The group name is namespace.schema.table, where namespace is the actual database name, schema is the actual schema name, and table is the actual table name.
2. For DB2, the group name will be like test_database.test_schema.test_table.

Tables Without primary keys #

Starting from version 3.4.0, DB2 CDC support tables that do not have a primary key. To use a table without primary keys, you must configure the scan.incremental.snapshot.chunk.key-column option and specify one non-null field.

There are two places that need to be taken care of.

1. If there is an index in the table, try to use a column which is contained in the index in scan.incremental.snapshot.chunk.key-column. This will increase the speed of select statement.
2. The processing semantics of a DB2 CDC table without primary keys is determined based on the behavior of the column that are specified by the scan.incremental.snapshot.chunk.key-column.

• If no update operation is performed on the specified column, the exactly-once semantics is ensured.
• If the update operation is performed on the specified column, only the at-least-once semantics is ensured. However, you can specify primary keys at downstream and perform the idempotence operation to ensure data correctness.

Warning #

Using a non-primary key column as the scan.incremental.snapshot.chunk.key-column for a DB2 table with primary keys may lead to data inconsistencies. Below is a scenario illustrating this issue and recommendations to mitigate potential problems.

Problem Scenario #

• Table Structure:

  • Primary Key: id
  • Chunk Key Column: pid (Not a primary key)

• Snapshot Splits:

  • Split 0: 1 < pid <= 3
  • Split 1: 3 < pid <= 5

• Operation:

  • Two different subtasks are reading Split 0 and Split 1 concurrently.
  • An update operation changes pid from 2 to 4 for id=0 while both splits are being read. This update occurs between the low and high watermark of both splits.

• Result:

  • Split 0: Contains the record [id=0, pid=2]
  • Split 1: Contains the record [id=0, pid=4]

Since the order of processing these records cannot be guaranteed, the final value of pid for id=0 may end up being either 2 or 4, leading to potential data inconsistencies.

Data Type Mapping #

Back to top