TICDC到kafka配置

1.软件下载

https://pingkai.cn/download#tidb-community

2.软件部署

以安装在/data下为例

cd /data
mkdir -p ticdc/{bin,conf,data,log,scripts}

tar -zvxf tidb-community-toolkit-v8.1.2-linux-amd64.tar.gz
cd tidb-community-toolkit-v8.1.2-linux-amd64/
tar -zvxf cdc-v8.1.2-linux-amd64.tar.gz
mv cdc ticdc/bin/

3.启动

touch /data/ticdc/conf/cdc.toml
cd /data/ticdc/scripts
#编辑启动脚本如下：
vi start_cdc.sh

#!/bin/bash
set -e
# WARNING: This file was auto-generated. Do not edit!
#          All your edit might be overwritten!
DEPLOY_DIR=/data/ticdc
cd "${DEPLOY_DIR}" || exit 1
exec bin/cdc server \
    --addr "0.0.0.0:8300" \
    --advertise-addr "192.168.2.56:8300" \
    --pd "http://192.168.2.53:2379" \
    --data-dir="/data/ticdc/data" \
    --gc-ttl 86400 \
    --config conf/cdc.toml \
    --log-file "/data/ticdc/log/cdc.log" 2>> "/data/ticdc/log/cdc_stderr.log"

说明：
--advertise-addr  为ticdc服务的地址，ip地址为部署TICDC的服务器
--pd 为源端tidb集群的pd节点，用逗号隔开。


nohup sh start_cdc.sh &

4.添加到kafka的changefeed

4.1 编辑配置文件

vi  chagefeed.conf
[sink]
dispatchers = [
{matcher = ['ticdc_test.*'], topic = "topic名", partition="index-value"},
]

说明：
matcher：配置抽取的表，*号为通配符
topic：定义kafka中topic的名称

4.2 添加changefeed命令

/data/ticdc/bin/cdc cli changefeed create   --pd=http://192.168.2.53:2379   --sink-uri=""kafka://192.168.2.56:9092/tidb-ticdc_test-t_order?protocol=canal-json""     --config="/data/ticdc/conf/chagefeed.conf"
--pd 源端tidb集群的pd节点，用逗号隔开
--sink-uri  kafka的连接方式

4.3 维护

#查看changefeed
/data/ticdc/bin/cdc cli changefeed list --server=http://192.168.2.56:8300
#停止changefeed
/data/ticdc/bin/cdc cli changefeed pause  --server=http://192.168.2.56:8300 --changefeed-id 3bb1c570-4e8d-4b50-9b3b-5876604bbf1f
#启动changefeed
/data/ticdc/bin/cdc cli changefeed resume  --server=http://192.168.2.56:8300 --changefeed-id 3bb1c570-4e8d-4b50-9b3b-5876604bbf1f
#更新changefeed配置
/data/ticdc/bin/cdc cli changefeed update  --server=http://192.168.2.56:8300 --changefeed-id 3bb1c570-4e8d-4b50-9b3b-5876604bbf1f  --config="/data/ticdc/conf/changefeed.toml"
#删除changefeed
/data/ticdc/bin/cdc cli changefeed remove  --server=http://192.168.2.56:8300 --changefeed-id 3bb1c570-4e8d-4b50-9b3b-5876604bbf1f
#查看changefeed细节
/data/ticdc/bin/cdc cli changefeed query --server=http://192.168.2.56:8300 --changefeed-id 3bb1c570-4e8d-4b50-9b3b-5876604bbf1f

5.kafka数据示例

#update:
{"id":0,"database":"ticdc_test","table":"t_order","pkNames":["order_id"],"isDdl":false,"type":"UPDATE","es":1766995442804,"ts":1766995443728,"sql":"","sqlType":{"order_id":3,"user_id":3,"order_date":93,"status":12,"total_amount":3},"mysqlType":{"order_id":"decimal","user_id":"decimal","order_date":"timestamp","status":"varchar","total_amount":"decimal"},"old":[{"order_id":"5000","user_id":"1","order_date":"2025-10-30 06:04:09","status":"ll2","total_amount":"12998"}],"data":[{"order_id":"5000","user_id":"1","order_date":"2025-10-30 06:04:09","status":"ll23","total_amount":"12998"}]}
#insert:
{"id":0,"database":"ticdc_test","table":"t_order","pkNames":["order_id"],"isDdl":false,"type":"DELETE","es":1766995571304,"ts":1766995572028,"sql":"","sqlType":{"order_id":3,"user_id":3,"order_date":93,"status":12,"total_amount":3},"mysqlType":{"order_id":"decimal","user_id":"decimal","order_date":"timestamp","status":"varchar","total_amount":"decimal"},"old":null,"data":[{"order_id":"5000","user_id":"1","order_date":"2025-10-30 06:04:09","status":"ll23","total_amount":"12998"}]}
#delete:
{"id":0,"database":"ticdc_test","table":"t_order","pkNames":["order_id"],"isDdl":false,"type":"DELETE","es":1766995571304,"ts":1766995572028,"sql":"","sqlType":{"order_id":3,"user_id":3,"order_date":93,"status":12,"total_amount":3},"mysqlType":{"order_id":"decimal","user_id":"decimal","order_date":"timestamp","status":"varchar","total_amount":"decimal"},"old":null,"data":[{"order_id":"5000","user_id":"1","order_date":"2025-10-30 06:04:09","status":"ll23","total_amount":"12998"}]}
#DDL:
{"id":0,"database":"ticdc_test","table":"t1","pkNames":null,"isDdl":true,"type":"ALTER","es":1766998832604,"ts":1766998833732,"sql":"ALTER TABLE `ticdc_test`.`t1` MODIFY COLUMN `name` VARCHAR(523) CHARACTER SET UTF8MB4 COLLATE utf8mb4_bin NULL DEFAULT NULL AFTER `id`","sqlType":null,"mysqlType":null,"data":null,"old":null}