TiDB CDC 连接器

TiDB CDC 连接器允许从 TiDB 数据库读取快照数据和增量数据。本文描述了如何设置 TiDB CDC 连接器来对 TiDB 数据库运行 SQL 查询。

依赖

为了设置 TiDB CDC 连接器，下表提供了使用构建自动化工具（如 Maven 或 SBT）和带有 SQL JAR 包的 SQL 客户端的两个项目的依赖关系信息。

Maven dependency

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-tidb-cdc</artifactId>
    <version>3.8-SNAPSHOT</version>
</dependency>

SQL Client JAR

下载链接仅适用于稳定版本。

下载 flink-sql-connector-tidb-cdc 到 <FLINK_HOME>/lib/ 目录下。

注意： 参考 flink-sql-connector-tidb-cdc 当前已发布的所有版本都可以在 Maven 中央仓库获取。

如何创建 TiDB CDC 表

TiDB CDC 表可以定义如下：

-- checkpoint every 3000 milliseconds                       
Flink SQL> SET 'execution.checkpointing.interval' = '3s';   

-- register a TiDB table 'orders' in Flink SQL
Flink SQL> CREATE TABLE orders (
     order_id INT,
     order_date TIMESTAMP(3),
     customer_name STRING,
     price DECIMAL(10, 5),
     product_id INT,
     order_status BOOLEAN,
     PRIMARY KEY(order_id) NOT ENFORCED
     ) WITH (
     'connector' = 'tidb-cdc',
     'tikv.grpc.timeout_in_ms' = '20000', 
     'pd-addresses' = 'localhost:2379',
     'database-name' = 'mydb',
     'table-name' = 'orders'
);
  
-- read snapshot and binlogs from orders table
Flink SQL> SELECT * FROM orders;

连接器配置项

    
        Option
        Required
        Default
        Type
        Description
      

    
      connector
      required
      (none)
      String
      指定要使用的连接器，这里应该是 'tidb-cdc'。
    

      database-name
      required
      (none)
      String
      要监视的 TiDB 服务器的数据库名称。
    

      table-name
      required
      (none)
      String
      需要监视的 TiDB 数据库的表名。
    

      scan.startup.mode
      optional
      initial
      String
      TiDB CDC 消费者可选的启动模式，合法的模式为 "initial" 和 "latest-offset"。
    

      pd-addresses
      required
      (none)
      String
      TiKV 集群的 PD 地址。
    

      host-mapping
      optional
      (none)
      String
      TiKV 集群的 host-mapping，用于配置公网 IP 和内网 IP 的映射。当 TiKV 集群运行在内网时，你可以将一组内网 IP 映射到公网 IP，以供外部的 Flink 集群访问。格式为 {Intranet IP1}:{Public IP1};{Intranet IP2}:{Public IP2}，例如 192.168.0.2:8.8.8.8;192.168.0.3:9.9.9.9。
    

      tikv.grpc.timeout_in_ms
      optional
      (none)
      Long
      TiKV GRPC 超时时间，单位为毫秒。
    

      tikv.grpc.scan_timeout_in_ms
      optional
      (none)
      Long
      TiKV GRPC 扫描超时时间，单位为毫秒。
    

      tikv.batch_get_concurrency
      optional
      20
      Integer
      TiKV GRPC batch get 的并发度。
    

      tikv.*
      optional
      (none)
      String
      透传 TiDB 客户端的属性。
    

Option	Required	Default	Type	Description
connector	required	(none)	String	指定要使用的连接器，这里应该是 `'tidb-cdc'`。
database-name	required	(none)	String	要监视的 TiDB 服务器的数据库名称。
table-name	required	(none)	String	需要监视的 TiDB 数据库的表名。
scan.startup.mode	optional	initial	String	TiDB CDC 消费者可选的启动模式，合法的模式为 "initial" 和 "latest-offset"。
pd-addresses	required	(none)	String	TiKV 集群的 PD 地址。
host-mapping	optional	(none)	String	TiKV 集群的 host-mapping，用于配置公网 IP 和内网 IP 的映射。当 TiKV 集群运行在内网时，你可以将一组内网 IP 映射到公网 IP，以供外部的 Flink 集群访问。格式为 {Intranet IP1}:{Public IP1};{Intranet IP2}:{Public IP2}，例如 192.168.0.2:8.8.8.8;192.168.0.3:9.9.9.9。
tikv.grpc.timeout_in_ms	optional	(none)	Long	TiKV GRPC 超时时间，单位为毫秒。
tikv.grpc.scan_timeout_in_ms	optional	(none)	Long	TiKV GRPC 扫描超时时间，单位为毫秒。
tikv.batch_get_concurrency	optional	20	Integer	TiKV GRPC batch get 的并发度。
tikv.*	optional	(none)	String	透传 TiDB 客户端的属性。

可用的元数据

下表中的元数据可以在 DDL 中作为只读（虚拟）meta 列声明。

Key	DataType	Description
table_name	STRING NOT NULL	当前记录所属的表名称。
database_name	STRING NOT NULL	当前记录所属的库名称。
op_ts	TIMESTAMP_LTZ(3) NOT NULL	当前记录表在数据库中更新的时间。如果从表的快照而不是 binlog 读取记录，该值将始终为 0。

下述创建表示例展示元数据列的用法：

CREATE TABLE products (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    table_name STRING METADATA  FROM 'table_name' VIRTUAL,
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    order_id INT,
    order_date TIMESTAMP(0),
    customer_name STRING,
    price DECIMAL(10, 5),
    product_id INT,
    order_status BOOLEAN,
    PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
    'connector' = 'tidb-cdc',
    'tikv.grpc.timeout_in_ms' = '20000',
    'pd-addresses' = 'localhost:2379',
    'database-name' = 'mydb',
    'table-name' = 'orders'
);

特性

Exactly-Once 处理

TiDB CDC 连接器是一个 Flink Source 连接器，它将首先读取数据库快照，然后继续读取变更事件，即使在处理时出现故障，也能准确读取数据。

启动模式

配置选项 scan.startup.mode 指定 TiDB CDC 使用者的启动模式。有效枚举包括：

initial（默认）：对捕获表的结构和数据执行快照；适用于需要获取捕获表数据完整表示的场景。
latest-offset：仅对捕获表的结构执行快照；适用于只需要获取从现在开始发生的变更的场景。

多线程读取

TiDB CDC source 可通过并行读取方式工作，因为有多个任务可以接收变更事件。

DataStream Source

TiDB CDC 连接器也可以是一个数据流源。你可以创建 SourceFunction，如下所示：

DataStream Source

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import org.apache.flink.cdc.connectors.tidb.TDBSourceOptions;
import org.apache.flink.cdc.connectors.tidb.TiDBSource;
import org.apache.flink.cdc.connectors.tidb.TiKVChangeEventDeserializationSchema;
import org.apache.flink.cdc.connectors.tidb.TiKVSnapshotEventDeserializationSchema;
import org.tikv.kvproto.Cdcpb;
import org.tikv.kvproto.Kvrpcpb;

import java.util.HashMap;

public class TiDBSourceExample {

    public static void main(String[] args) throws Exception {

        SourceFunction<String> tidbSource =
            TiDBSource.<String>builder()
                .database("mydb") // set captured database
                .tableName("products") // set captured table
                .tiConf(
                    TDBSourceOptions.getTiConfiguration(
                        "localhost:2399", new HashMap<>()))
                .snapshotEventDeserializer(
                    new TiKVSnapshotEventDeserializationSchema<String>() {
                        @Override
                        public void deserialize(
                            Kvrpcpb.KvPair record, Collector<String> out)
                            throws Exception {
                            out.collect(record.toString());
                        }

                        @Override
                        public TypeInformation<String> getProducedType() {
                            return BasicTypeInfo.STRING_TYPE_INFO;
                        }
                    })
                .changeEventDeserializer(
                    new TiKVChangeEventDeserializationSchema<String>() {
                        @Override
                        public void deserialize(
                            Cdcpb.Event.Row record, Collector<String> out)
                            throws Exception {
                            out.collect(record.toString());
                        }

                        @Override
                        public TypeInformation<String> getProducedType() {
                            return BasicTypeInfo.STRING_TYPE_INFO;
                        }
                    })
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // enable checkpoint
        env.enableCheckpointing(3000);
        env.addSource(tidbSource).print().setParallelism(1);

        env.execute("Print TiDB Snapshot + Binlog");
    }
}

数据类型映射

TiDB type	Flink SQL type	NOTE
TINYINT	TINYINT
SMALLINT TINYINT UNSIGNED	SMALLINT
INT MEDIUMINT SMALLINT UNSIGNED	INT
BIGINT INT UNSIGNED	BIGINT
BIGINT UNSIGNED	DECIMAL(20, 0)
FLOAT	FLOAT
REAL DOUBLE	DOUBLE
NUMERIC(p, s) DECIMAL(p, s) where p <= 38	DECIMAL(p, s)
NUMERIC(p, s) DECIMAL(p, s) where 38 < p <= 65	STRING	在 TiDB 中，十进制数据类型的精度高达 65，但在 Flink 中，十进制数据类型的精度仅限于 38。所以，如果定义精度大于 38 的十进制列，则应将其映射到字符串以避免精度损失。
BOOLEAN TINYINT(1) BIT(1)	BOOLEAN
DATE	DATE
TIME [(p)]	TIME [(p)]
TIMESTAMP [(p)]	TIMESTAMP_LTZ [(p)]
DATETIME [(p)]	TIMESTAMP [(p)]
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)
BIT(n)	BINARY(⌈n/8⌉)
BINARY(n)	BINARY(n)
TINYTEXT TEXT MEDIUMTEXT LONGTEXT	STRING
TINYBLOB BLOB MEDIUMBLOB LONGBLOB	BYTES	目前，对于 TiDB 中的 BLOB 数据类型，仅支持长度不大于 2,147,483,647(2 ** 31 - 1) 的 blob。
YEAR	INT
ENUM	STRING
JSON	STRING	JSON 数据类型将在 Flink 中转换为 JSON 格式的字符串。
SET	ARRAY<STRING>	因为 TiDB 中的 SET 数据类型是一个字符串对象，可以有零个或多个值，它应该始终映射到字符串数组。

依赖​

Maven dependency​

SQL Client JAR​

如何创建 TiDB CDC 表​

连接器配置项​

可用的元数据​

特性​

Exactly-Once 处理​

启动模式​

多线程读取​

DataStream Source​

DataStream Source​

数据类型映射​

依赖

Maven dependency

SQL Client JAR

如何创建 TiDB CDC 表

连接器配置项

可用的元数据

特性

Exactly-Once 处理

启动模式

多线程读取

DataStream Source

DataStream Source

数据类型映射