Skip to main content
Version: Next

DB2 CDC

DB2 CDC source connector

Support DB2 Version​

  • DB2 LUW 11.5 or later versions supported by Debezium DB2 connector

Support Those Engines​

SeaTunnel Zeta
Flink

Key Features​

Description​

The DB2 CDC connector reads snapshot data and incremental data from DB2 tables that have been put into capture mode. It uses Debezium DB2 internally and can continue streaming committed INSERT, UPDATE and DELETE changes after the initial snapshot is finished.

Supported DataSource Info​

DatasourceSupported versionsDriverUrlMaven
DB2DB2 LUW 11.5 or later versions supported by Debezium DB2 connectorcom.ibm.db2.jcc.DB2Driverjdbc:db2://127.0.0.1:50000/testdbhttps://mvnrepository.com/artifact/com.ibm.db2.jcc/db2jcc

Using Dependency​

Install Jdbc Driver​

  1. You need to ensure that the DB2 JDBC driver jar package has been placed in directory ${SEATUNNEL_HOME}/plugins/.

For SeaTunnel Zeta Engine​

  1. You need to ensure that the DB2 JDBC driver jar package has been placed in directory ${SEATUNNEL_HOME}/lib/.

Data Type Mapping​

DB2 Data TypeSeaTunnel Data Type
BOOLEANBOOLEAN
SMALLINTSHORT
INT
INTEGER		INT
BIGINTBIGINT
DECIMAL
DEC
NUMERIC
NUM		DECIMAL
REALFLOAT
DOUBLE
DECFLOAT		DOUBLE
CHAR
CHARACTER
VARCHAR
LONG VARCHAR
CLOB
GRAPHIC
VARGRAPHIC
DBCLOB
XML		STRING
BINARY
VARBINARY
BLOB		BYTES
DATEDATE
TIMETIME
TIMESTAMPTIMESTAMP

Source Options​

NameTypeRequiredDefaultDescription
usernameStringYes-User name used to connect to DB2.
passwordStringYes-Password used to connect to DB2.
urlStringYes-DB2 JDBC URL. The URL must include a database name, for example jdbc:db2://127.0.0.1:50000/testdb.
database-namesListNoThe database parsed from urlDatabase name to monitor. DB2 CDC captures one database in one source.
table-namesListYes when table-pattern is not set-Table names to monitor, using databaseName.schemaName.tableName, for example testdb.DB2INST1.CUSTOMERS.
table-patternStringYes when table-names is not set-Regular expression used to discover captured tables.
table-names-configListNo-Table config list. For example: [{"table": "testdb.DB2INST1.CUSTOMERS","primaryKeys": ["ID"],"snapshotSplitColumn": "ID"}].
startup.modeEnumNoINITIALOptional startup mode for DB2 CDC source. Valid values are initial, earliest and latest.
stop.modeEnumNoNEVEROptional stop mode for DB2 CDC source. Valid value is never.
incremental.parallelismIntegerNo1The number of parallel readers in the incremental phase.
snapshot.split.sizeIntegerNo8096The split size of table snapshot.
snapshot.fetch.sizeIntegerNo1024The maximum fetch size for each poll when reading table snapshot.
server-time-zoneStringNoUTCThe session time zone in database server.
connect.timeout.msDurationNo30sThe maximum time that the connector should wait after trying to connect to the database server before timing out.
connect.max-retriesIntegerNo3The maximum retry times to build database server connection.
connection.pool.sizeIntegerNo20The connection pool size.
chunk-key.even-distribution.factor.upper-boundDoubleNo100The upper bound used to decide whether a split key is evenly distributed.
chunk-key.even-distribution.factor.lower-boundDoubleNo0.05The lower bound used to decide whether a split key is evenly distributed.
sample-sharding.thresholdintNo1000The estimated shard count threshold that triggers sample-based sharding for unevenly distributed split keys.
inverse-sampling.rateintNo1000The inverse sampling rate used by sample-based sharding.
exactly_onceBooleanNofalseEnable exactly-once semantics for initial snapshot handoff.
debezium.*configNo-Pass-through Debezium DB2 connector properties.
formatEnumNoDEFAULTOptional output format. Valid values are DEFAULT and COMPATIBLE_DEBEZIUM_JSON.
common-optionsNo-Source plugin common parameters, please refer to Source Common Options for details.

Enable DB2 CDC​

DB2 CDC depends on DB2 SQL replication and ASN capture tables. Verify the required IBM replication license for your environment before enabling capture. The database administrator must put every source table into capture mode before running SeaTunnel. You can use DB2 control commands or Debezium's management UDFs. The following commands show the common UDF workflow:

VALUES ASNCDC.ASNCDCSERVICES('status','asncdc');
VALUES ASNCDC.ASNCDCSERVICES('start','asncdc');
CALL ASNCDC.ADDTABLE('DB2INST1', 'CUSTOMERS');
VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');

For complete DB2 server setup, permissions and ASN capture agent configuration, refer to the Debezium DB2 connector setup guide.

Task Example​

Initial Read Simple​

This example reads an initial snapshot and then continues to read incremental changes.

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 5000
}

source {
  DB2-CDC {
    plugin_output = "customers"
    username = "db2inst1"
    password = "db2inst1"
    startup.mode = "initial"
    database-names = ["testdb"]
    table-names = ["testdb.DB2INST1.CUSTOMERS"]
    url = "jdbc:db2://127.0.0.1:50000/testdb"
  }
}

sink {
  console {
    plugin_input = "customers"
  }
}

Incremental Read Simple​

This example starts from the latest DB2 LSN and prints newly changed data.

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 5000
}

source {
  DB2-CDC {
    plugin_output = "customers"
    username = "db2inst1"
    password = "db2inst1"
    startup.mode = "latest"
    database-names = ["testdb"]
    table-names = ["testdb.DB2INST1.CUSTOMERS"]
    url = "jdbc:db2://127.0.0.1:50000/testdb"
  }
}

sink {
  console {
    plugin_input = "customers"
  }
}

Support Custom Primary Key For Table​

source {
  DB2-CDC {
    plugin_output = "customers"
    username = "db2inst1"
    password = "db2inst1"
    startup.mode = "initial"
    database-names = ["testdb"]
    table-names = ["testdb.DB2INST1.CUSTOMERS"]
    table-names-config = [
      {
        table = "testdb.DB2INST1.CUSTOMERS"
        primaryKeys = ["ID"]
        snapshotSplitColumn = "ID"
      }
    ]
    url = "jdbc:db2://127.0.0.1:50000/testdb"
  }
}

Changelog​

Change Log
ChangeCommitVersion