tidb实时同步到mysql

news/2025/2/22 18:08:59

客户要求实时同步表的数据到mysql，但这个表在tidb。
测试直接通过tidb cdc写入到mysql，有些字段是null，所以中间加了一个kafka实现

客户库中创建表

CREATE TABLE tb_1 (
    id bigint primary key,
    cid bigint,
    gid bigint,
    fee DECIMAL(10,2),
    created_at timestamp,
    type smallint,
    remark string ,
    key i_cid(cid)
);

tidbcdckafka_18">tidb配置cdc，写入到kafka

1、tidb添加cdc组件
2、配置cdc任务
cat your.toml

case-sensitive = true
enable-old-value = true
[filter]
rules = ['db.tb_1']

3、启动任务

tiup ctl:v5.3.0 cdc changefeed create --pd=http://pd-ip:2379 \
--sink-uri="kafka://kafka-ip:9092/your-topic?kafka-version=1.1.1&partition-num=1&max-message-bytes=67108864&replication-factor=1&protocol=canal-json" --changefeed-id="my" --sort-engine="unified" \
--start-ts=453870757254529193 --config your.toml

protocol=canal-json 使用这个格式
–start-ts这个通过一次导出查看，cat db/metadata

dumpling -u root -p pwd -h tidb-ip -P 3306  -F 1GiB --compress gzip -t 2 -o db -B db -T db.tb --where "cid=123456"

4、更新一条数据，看看kafka是不是有了

mysql_43">创建SQL作业，从kafka消费后入mysql

CREATE TABLE tb_1 (
    id bigint,
    cid bigint,
    gid bigint,
    fee DECIMAL(10,2),
    create_time timestamp,
    type smallint,
    remark string,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  -- 定义 Kafka 参数
  'connector' = 'kafka',
  'topic' = 'your-topic',
  'scan.startup.mode' = 'latest-offset', 
  'properties.bootstrap.servers' = 'ip:9092',
  'properties.group.id' = 'your-group',
  'format' = 'canal-json', -- tidb 支持该方式
  'canal-json.ignore-parse-errors' = 'false'
);



CREATE TABLE kh_tb_1 (
    id bigint,
    cid bigint,
    gid bigint,
    fee DECIMAL(10,2),
    create_time timestamp,
    type smallint,
    remark string,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://xxxxxx:3306/db?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai',
  'table-name' = 'tb_1',
  'username' = 'user',
  'password' = 'pwd',
  'sink.buffer-flush.max-rows' = '5000',
  'sink.buffer-flush.interval' = '2s',
  'sink.max-retries' = '10'
);

insert into kh_tb_1 select * from tb_1 where cid=xxxxxxx;