Flink TiDB CDC 环境配置与验证

一、TiDB 数据库核心配置

1. 启用 TiCDC 服务

确保 TiDB 集群已部署 TiCDC 组件（版本需兼容 Flink CDC 3.0.1），并启动同步服务：

# 示例：启动 TiCDC 捕获 changefeed
cdc cli changefeed create \
  --pd="localhost:2379" \
  --sink-uri="blackhole://" \
  --changefeed-id="flink-cdc-demo"

2. 验证 PD 地址

获取 TiDB 集群的 PD（Placement Driver）地址，用于 Flink 连接：

# 查询 PD 地址
cat /path/to/tidb-deploy/pd-2379/conf/config.toml | grep advertise-client-urls
# 输出示例：advertise-client-urls = "http://192.168.1.10:2379"

二、Flink 环境集成配置

1. 添加Maven依赖

<!-- TiDB CDC 连接器 -->
<dependency>
    <groupId>com.ververica</groupId>
    <artifactId>flink-sql-connector-tidb-cdc</artifactId>
    <version>3.0.1</version>
    <scope>provided</scope>
</dependency>

<!-- TiDB JDBC 驱动（若需要） -->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.26</version>
</dependency>

2. SQL Client部署

1. 下载 TiDB CDC 连接器 JAR：
   flink-sql-connector-tidb-cdc-3.0.1.jar
2. 将 JAR 包放入 $FLINK_HOME/lib/ 后重启 Flink 集群。

三、Flink SQL 表定义与参数详解

1. 完整建表示例（含元数据）

-- 配置 checkpoint（每 3 秒）
SET 'execution.checkpointing.interval' = '3s';

-- 创建 TiDB CDC 表
CREATE TABLE tidb_orders (
    order_id INT,
    order_date TIMESTAMP(3),
    customer_name STRING,
    price DECIMAL(10, 5),
    order_status BOOLEAN,
    -- 元数据列
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    table_name STRING METADATA FROM 'table_name' VIRTUAL,
    op_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
    'connector' = 'tidb-cdc',
    'pd-addresses' = '192.168.1.10:2379',  -- PD 地址（必填）
    'database-name' = 'mydb',
    'table-name' = 'orders',
    'scan.startup.mode' = 'initial',  -- 启动模式：initial（全量+增量）或 latest-offset（仅增量）
    'tikv.grpc.timeout_in_ms' = '20000',  -- GRPC 超时时间（毫秒）
    'tikv.batch_get_concurrency' = '20'  -- 批量获取并发度
);

2. 核心参数详解

四、环境验证与测试

1. 准备测试数据（TiDB）

-- 连接 TiDB
mysql -h 127.0.0.1 -P 4000 -u root

-- 创建测试数据库和表
CREATE DATABASE mydb;
USE mydb;

CREATE TABLE orders (
    order_id INT PRIMARY KEY,
    order_date TIMESTAMP(3),
    customer_name VARCHAR(100),
    price DECIMAL(10, 5),
    order_status BOOLEAN
);

-- 插入测试数据
INSERT INTO orders VALUES 
(1, '2023-01-01 10:00:00.000', 'Alice', 100.50, true),
(2, '2023-01-02 11:00:00.000', 'Bob', 200.75, false);
COMMIT;

2. Flink SQL 验证

-- 查询 TiDB CDC 表（首次触发全量扫描）
SELECT * FROM tidb_orders;

-- 在 TiDB 中更新数据
UPDATE mydb.orders SET price = 150.00 WHERE order_id = 1;
COMMIT;

-- 观察 Flink 输出：应显示更新后的记录，op_ts 为变更时间

3. DataStream API 验证

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
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.apache.flink.cdc.connectors.tidb.TDBSourceOptions;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.util.Collector;
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 {
        // 配置 TiDB Source
        TiDBSource<String> source = TiDBSource.<String>builder()
                .database("mydb")
                .tableName("orders")
                .tiConf(TDBSourceOptions.getTiConfiguration("192.168.1.10:2379", new HashMap<>()))
                .snapshotEventDeserializer(new TiKVSnapshotEventDeserializationSchema<String>() {
                    @Override
                    public void deserialize(Kvrpcpb.KvPair record, Collector<String> out) {
                        out.collect("Snapshot: " + 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) {
                        out.collect("Change: " + record.toString());
                    }
                    @Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;
                    }
                })
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(3000);
        env.addSource(source).print().setParallelism(1);
        env.execute("TiDB CDC Example");
    }
}

五、常见问题与解决方案

1. 连接 PD 失败

   ERROR: Failed to connect to PD server at 192.168.1.10:2379
   

   • 解决方案：
     • 确认 PD 地址和端口是否正确（可通过 ping 和 telnet 验证）
     • 检查 TiDB 集群状态：cdc cli cluster info --pd=http://192.168.1.10:2379

2. 权限不足

   ERROR: Access denied for user 'root'@'%'
   

   • 解决方案：
     • 确保 Flink 连接 TiDB 的用户有查询权限
     • 示例授权：GRANT SELECT ON mydb.orders TO 'flink_user'@'%';

3. 大数类型精度丢失

   • 问题：TiDB 中 DECIMAL(65, 0) 映射到 Flink 时可能截断
   • 解决方案：

     -- 将超高精度 DECIMAL 映射为 STRING
     CREATE TABLE tidb_orders (
         -- 其他字段...
         big_amount STRING  -- 替代 DECIMAL(65, 0)
     ) WITH (...)
     

4. TiCDC 同步延迟

   • 解决方案：
     • 调整 tikv.grpc.timeout_in_ms 增大超时时间
     • 增加 tikv.batch_get_concurrency 提高并发度

六、生产环境优化建议

1. 性能调优

   • 增大批量处理大小：

     'tikv.scan_batch_size' = '1000',  -- 单次扫描行数
     'tikv.scan_concurrency' = '16'    -- 扫描并发度
     

2. 高可用配置

   • 配置多个 PD 地址提高可用性：

     'pd-addresses' = 'pd1:2379,pd2:2379,pd3:2379'
     

3. 监控与维护

   • 监控 TiCDC 同步状态：

     cdc cli changefeed list --pd=http://pd:2379
     

   • 定期清理 TiCDC 历史日志：

     cdc cli gc --pd=http://pd:2379 --older-than=7d
     

通过以上步骤，可完成 Flink TiDB CDC 的全流程配置与验证。生产环境中需特别注意 PD 地址配置、大数类型映射及 TiCDC 服务稳定性，以确保数据一致性和系统性能。