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.

由于 Db2 Connector 采用的 IPLA 协议与 Flink CDC 项目不兼容，我们无法在 jar 包中提供 Db2 连接器。 您可能需要手动配置以下依赖：

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");  } } 

可用的指标 #

指标系统能够帮助了解分片分发的进展， 下面列举出了支持的 Flink 指标 Flink metrics:

注意:

1. Group 名称是 namespace.schema.table，这里的 namespace 是实际的数据库名称， schema 是实际的 schema 名称， table 是实际的表名称。
2. 对于 DB2，Group 的名称会类似于 test_database.test_schema.test_table。

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");  } } 

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