import ChangeLog from ‘../changelog/connector-cdc-tidb.md’;

TiDB CDC

TiDB CDC source connector

Support Those Engines

SeaTunnel Zeta
Flink

Key features

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

Description

The TiDB CDC connector allows for reading snapshot data and incremental data from TiDB database. This document describes how to set up the TiDB CDC connector to snapshot data and capture streaming event in TiDB database.

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 has 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 has been placed in directory ${SEATUNNEL_HOME}/lib/.

Please download and put Mysql driver and tikv-java-client in ${SEATUNNEL_HOME}/lib/ dir. For example: cp mysql-connector-java-xxx.jar $SEATUNNEL_HOME/lib/

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 ENUM	STRING
DATE	DATE
TIME(s)	TIME(s)
DATETIME TIMESTAMP(s)	TIMESTAMP(s)
BINARY VARBINAR BIT(p) TINYBLOB MEDIUMBLOB BLOB LONGBLOB GEOMETRY	BYTES

Source Options

Name	Type	Required	Default	Description
url	String	Yes	-	The URL of the JDBC connection. Refer to a case: jdbc:mysql://tidb0:4000/inventory.
username	String	Yes	-	Name of the database to use when connecting to the database server.
password	String	Yes	-	Password to use when connecting to the database server.
pd-addresses	String	Yes	-	TiKV cluster's PD address
database-name	String	Yes	-	Database name of the database to monitor.
table-name	String	Yes	-	Table name of the database to monitor. The table name needs to include the database name.
startup.mode	Enum	No	INITIAL	Optional startup mode for TiDB CDC consumer, valid enumerations are initial, earliest, latest and specific. initial: Synchronize historical data at startup, and then synchronize incremental data. earliest: Startup from the earliest offset possible. latest: Startup from the latest offset. specific: Startup from user-supplied specific offsets.
batch-size-per-scan	Int	No	1000	Size per scan.
tikv.grpc.timeout_in_ms	Long	No	-	TiKV GRPC timeout in ms.
tikv.grpc.scan_timeout_in_ms	Long	No	-	TiKV GRPC scan timeout in ms.
tikv.batch_get_concurrency	Integer	No	-	TiKV GRPC batch get concurrency
tikv.batch_scan_concurrency	Integer	No	-	TiKV GRPC batch scan concurrency

Task Example

Simple

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

source {
  # This is a example source plugin **only for test and demonstrate the feature source plugin**
  TiDB-CDC {
    plugin_output = "products_tidb_cdc"
    url = "jdbc:mysql://tidb0:4000/inventory"
    driver = "com.mysql.cj.jdbc.Driver"
    tikv.grpc.timeout_in_ms = 20000
    pd-addresses = "pd0:2379"
    username = "root"
    password = ""
    database-name = "inventory"
    table-name = "products"
  }
}

transform {
}

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

Changelog