TICDC同步数据到kafka

配置文件

tidb2kafka.toml

case-sensitive = true

enable-old-value = true

[filter]
ignore-txn-start-ts = [1, 2]


rules = [
'my_test_schema.my_test_table01',
'my_test_schema.my_test_table02',
'my_test_schema.my_test_table03',
'my_test_schema.my_test_table04']

[mounter]
worker-num = 4


[[sink.dispatchers]]
matcher = [
'my_test_schema.my_test_table01',
'my_test_schema.my_test_table02',
'my_test_schema.my_test_table03',
'my_test_schema.my_test_table04']
dispatcher = 'ts'


[sink]
protocol  = "canal-json"

创建任务

cdc cli changefeed create --pd=http://10.10.10.105:2379 --sink-uri="kafka://10.10.10.231:9092/my_test_topic?protocol=canal-json&kafka-version=2.6.0&partition-num=10&max-message-bytes=67108864&replication-factor=3" --changefeed-id="my_test-to-kafka-dev" --config="tidb2kafka.toml" --start-ts="0"

暂停任务

cdc cli changefeed pause --pd=http://10.10.10.105:2379 --changefeed-id="my_test-to-kafka-dev"

查询任务

cdc cli changefeed query -s --pd=http://10.10.10.105:2379 --changefeed-id="my_test-to-kafka-dev"

删除任务

cdc cli changefeed remove -s --pd=http://10.10.10.105:2379 --changefeed-id="my_test-to-kafka-dev"

更新任务

cdc cli changefeed update--pd=http://10.10.10.105:2379 --sink-uri="kafka://10.10.10.231:9092/my_test_db_sync?protocol=canal-json&kafka-version=2.6.0&partition-num=10&max-message-bytes=67108864&replication-factor=3" --changefeed-id="dev-my_test-to-kafka" --config="tidb2kafka.toml"

继续任务

cdc cli changefeed resume -s --pd=http://10.10.10.105:2379 --changefeed-id="my_test-to-kafka-dev"

查看进程

cdc cli processor list  --pd=http://10.10.10.105:2379  --changefeed-id="my_test-to-kafka-dev"