TiDB CDC

TiDB CDC source connector

Support Those Engines

SeaTunnel Zeta
Flink

Description

The TiDB CDC connector reads snapshot data and incremental change events from TiDB by talking to its TiKV placement-driver (PD) and TiKV nodes through the tikv-client-java Java client. It supports parallel snapshot reads and exactly-once streaming, and is the recommended way to bring TiDB tables into a SeaTunnel pipeline.

Supported DataSource Info

Datasource	Supported versions	Driver	Url	Maven
MySQL	MySQL: 5.5, 5.6, 5.7, 8.0.x RDS MySQL: 5.6, 5.7, 8.0.x	com.mysql.cj.jdbc.Driver	jdbc:mysql://localhost:3306/test	https://mvnrepository.com/artifact/mysql/mysql-connector-java/8.0.28
tikv-client-java	3.2.0	-	-	https://mvnrepository.com/artifact/org.tikv/tikv-client-java/3.2.0

Using Dependency

Install JDBC Driver

For Flink Engine

1. You need to ensure that the jdbc driver jar package and the tikv-client-java jar package have been placed in directory ${SEATUNNEL_HOME}/plugins/.

For SeaTunnel Zeta Engine

1. You need to ensure that the jdbc driver jar package and the tikv-client-java jar package have been placed in directory ${SEATUNNEL_HOME}/lib/.

Please download and put the MySQL driver and tikv-client-java in the directory required by your engine.

Key features

• batch
• stream
• exactly-once
• column projection
• parallelism
• support user-defined split

Data Type Mapping

MySQL Data Type	SeaTunnel Data Type
BIT(1) TINYINT(1)	BOOLEAN
TINYINT	TINYINT
TINYINT UNSIGNED SMALLINT	SMALLINT
SMALLINT UNSIGNED MEDIUMINT MEDIUMINT UNSIGNED INT INTEGER YEAR	INT
INT UNSIGNED INTEGER UNSIGNED BIGINT	BIGINT
BIGINT UNSIGNED	DECIMAL(20, 0)
DECIMAL(p, s) DECIMAL(p, s) UNSIGNED NUMERIC(p, s) NUMERIC(p, s) UNSIGNED	DECIMAL(p, s)
FLOAT FLOAT UNSIGNED	FLOAT
DOUBLE DOUBLE UNSIGNED REAL REAL UNSIGNED	DOUBLE
CHAR VARCHAR TINYTEXT MEDIUMTEXT TEXT LONGTEXT ENUM JSON	STRING
DATE	DATE
TIME(s)	TIME(s)
DATETIME TIMESTAMP(s)	TIMESTAMP(s)
BINARY VARBINARY BIT(p) TINYBLOB MEDIUMBLOB BLOB LONGBLOB GEOMETRY	BYTES

Source Options

Name	Type	Required	Default	Description
url	String	Yes	-	MySQL-compatible JDBC URL used to discover table metadata. Example: jdbc:mysql://tidb0:4000/inventory.
username	String	Yes	-	Username used to connect to the TiDB server.
password	String	Yes	-	Password used to connect to the TiDB server.
pd-addresses	String	Yes	-	TiKV placement-driver (PD) endpoints, comma-separated, e.g. pd0:2379,pd1:2379.
database-name	String	Yes	-	Name of the TiDB database to monitor.
table-name	String	Yes	-	Table name to monitor inside database-name. Do not include the database name.
startup.mode	Enum	No	INITIAL	Optional startup mode for the TiDB CDC consumer. Valid values are initial, earliest, latest. initial snapshots historical data first, then keeps reading incremental changes. earliest starts from the earliest available offset. latest skips the initial snapshot and only consumes new changes from now on.
batch-size-per-scan	Int	No	1000	Number of rows fetched per scan request against TiKV.
tikv.grpc.timeout_in_ms	Long	No	-	TiKV gRPC client timeout in milliseconds. Increase it when TiKV is slow to respond under load.
tikv.grpc.scan_timeout_in_ms	Long	No	-	TiKV gRPC scan timeout in milliseconds. Increase it when large scans time out.
tikv.batch_get_concurrency	Integer	No	-	Concurrency for TiKV BatchGet requests. Tune upward when reads are bottlenecked by TiKV CPU.
tikv.batch_scan_concurrency	Integer	No	-	Concurrency for TiKV BatchScan requests. Tune upward when snapshot reads are bottlenecked by TiKV CPU.

Task Example

Simple

This example streams CDC events from a TiDB table into a JDBC sink. Set job.mode = "STREAMING" and a checkpoint interval so incremental events flow continuously.

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

source {
  TiDB-CDC {
    plugin_output = "products_tidb_cdc"
    url = "jdbc:mysql://tidb0:4000/tidb_cdc"
    driver = "com.mysql.cj.jdbc.Driver"
    tikv.grpc.timeout_in_ms = 20000
    pd-addresses = "pd0:2379"
    username = "root"
    password = ""
    database-name = "tidb_cdc"
    table-name = "tidb_cdc_e2e_source_table"
  }
}

sink {
  Jdbc {
    plugin_input = "products_tidb_cdc"
    url = "jdbc:mysql://tidb0:4000/tidb_cdc"
    driver = "com.mysql.cj.jdbc.Driver"
    username = "root"
    password = ""
    database = "tidb_cdc"
    table = "tidb_cdc_e2e_sink_table"
    generate_sink_sql = true
    primary_keys = ["id"]
  }
}

Start From Latest Offset

Use startup.mode = "latest" when only new changes are needed and the historical snapshot should be skipped.

source {
  TiDB-CDC {
    url = "jdbc:mysql://tidb0:4000/tidb_cdc"
    driver = "com.mysql.cj.jdbc.Driver"
    pd-addresses = "pd0:2379"
    username = "root"
    password = ""
    database-name = "tidb_cdc"
    table-name = "tidb_cdc_e2e_source_table"
    startup.mode = "latest"
  }
}

Notes

• TiDB CDC reads one table per source block. Use multiple TiDB-CDC source blocks if one job needs to capture multiple tables.
• startup.mode = "specific" is not a valid TiDB CDC option. Use initial, earliest, or latest.
• Tune tikv.grpc.* and tikv.batch_*_concurrency only when the default TiKV client settings are not enough for your cluster.

Changelog

Change Log

Change	Commit	Version
[Feature][Core] Add plugin directory support for each connector (#9650)	https://github.com/apache/seatunnel/commit/4beb2b9336	2.3.12
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)	https://github.com/apache/seatunnel/commit/9212a77140	2.3.12
[Feature][Connectors-v2] Optimize the size of CDC JAR Files (#9546)	https://github.com/apache/seatunnel/commit/1dd19c6823	2.3.12
[Fix][Connector-V2] Correct typo in batch-size-per-scan option key (#9434)	https://github.com/apache/seatunnel/commit/6cf258127f	2.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)	https://github.com/apache/seatunnel/commit/4f5adeb1c7	2.3.11
[Feature] Support tidb cdc connector source #7199 (#7477)	https://github.com/apache/seatunnel/commit/87ec786bd6	2.3.8