作者： lqbyz 原文来源：https://tidb.net/blog/ff66721d
使用 TiCDC 将 TiDB test 数据库多张表以 AVRO 格式发送到 Kafka 多个 Topic ，然后使用 Confluent 自带开源 connect 将 Kafka 多个 topic 数据实时写入到 Oracle 数据库，此链路支持实时数据 insert/delete/update/create table ddl/add column ddl 等。理论上此链路还可以支持下游为其它异构数据库。
一、通过 TiCDC 同步数据到 kafka
1、通过 tiup 安装 ticdc
1、编辑扩展文件[root@172-16-5-146 tidb]# cat cdc.yamlcdc_servers: - host: 172.16.6.132 gc-ttl: 86400 data_dir: "/home/cdc-data" 2、扩展该文件[root@172-16-5-146 tidb]# tiup cluster scale-out tidb-test cdc.yaml -uroot -p
3、查看该配置文件[root@172-16-5-146 tidb]# tiup cluster display tidb-testtiup is checking updates for component cluster ...A new version of cluster is available: The latest version: v1.10.3 Local installed version: v1.10.2 Update current component: tiup update cluster Update all components: tiup update --all
Starting component `cluster`: /root/.tiup/components/cluster/v1.10.2/tiup-cluster display tidb-testCluster type: tidbCluster name: tidb-testCluster version: v6.1.0Deploy user: tidbSSH type: builtinDashboard URL: http://172.16.5.146:2399/dashboardGrafana URL: http://172.16.5.146:3001ID Role Host Ports OS/Arch Status Data Dir Deploy Dir-- ---- ---- ----- ------- ------ -------- ----------172.16.5.146:9099 alertmanager 172.16.5.146 9099/9095 linux/x86_64 Up /home/tidb/tidb-deploy/alertmanager-9099/data /home/tidb/tidb-deploy/alertmanager-9099172.16.6.132:8300 cdc 172.16.6.132 8300 linux/x86_64 Up /home/cdc-data /home/tidb/tidb-deploy/cdc-8300
复制代码
2、安装 Confluent 环境
2.1、下载安装文件
登陆 Confluent 官网： https://www.confluent.io/get-started/?product=software
下载 confluent-7.2.1.tar ，将下载的文件传输到服务器，解压缩为 /root/software/confluent-7.2.1
[root@172-16-6-132 kafka]# lltotal 1952776-rw-r--r-- 1 root root 104 Sep 21 13:50 c.confdrwxr-xr-x 8 centos centos 92 Jul 14 12:07 confluent-7.2.1-rw-r--r-- 1 root root 1999633201 Sep 21 10:33 confluent-7.2.1.tar.gz-rw-r--r-- 1 root root 672 Sep 21 18:08 kafka.conf[root@172-16-6-132 kafka]# du -sh *4.0K c.conf2.3G confluent-7.2.11.9G confluent-7.2.1.tar.gz4.0K kafka.conf
复制代码
2.2、配置环境变量
[root@172-16-6-132 kafka]#vim ~/.bashrc# .bashrc
# User specific aliases and functions
alias rm='rm -i'alias cp='cp -i'alias mv='mv -i'
# Source global definitionsif [ -f /etc/bashrc ]; then . /etc/bashrcfi###最下边添加如下：export CONFLUENT_HOME=/home/kafka/confluent-7.2.1export PATH=$PATH:$CONFLUENT_HOME/bin[root@172-16-6-132 kafka]# source ~/.bashrc
复制代码
2.3、安装 Confluent
cd /home/kafka/confluent-7.2.1/bin./confluent-hub install --no-prompt confluentinc/kafka-connect-datagen:latest
复制代码
2.4、启动 Confluent
停止confluent local services stop启动confluent local services start查看状态confluent local services status
##如果有启动某些组件失败的时候，需要指定组件重新启动查看某个组件的日志confluent local services schema-registry log -f重新启动某个组件confluent local services schema-registry start查看某个组件的状态confluent local services schema-registry status
复制代码
2.5、Confluent 组件介绍
[root@172-16-6-132 bin]# confluent local servicesManage Confluent Platform services.
Usage: confluent local services [command]
Available Commands: connect Manage Connect. control-center Manage Control Center. kafka Manage Apache Kafka®. kafka-rest Manage Kafka REST. ksql-server Manage ksqlDB Server. list List all Confluent Platform services. schema-registry Manage Schema Registry. start Start all Confluent Platform services. status Check the status of all Confluent Platform services. stop Stop all Confluent Platform services. top View resource usage for all Confluent Platform services. zookeeper Manage Apache ZooKeeper™.
Global Flags: -h, --help Show help for this command. -v, --verbose count Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).
Use "confluent local services [command] --help" for more information about a command.
复制代码
Connect 是本文后续使用的组件，它支持从 kafka 消费数据写入到下游组件，下游组件可以通过插件支持各种异构存储系统如：Oracle、MySQL、SQL Server、ES 等
Control Center 是 Web 控制页面，用户可以在页面上管理 Kafka 集群和数据，管理 connect 任务和插件
Kafka 消息组件，不做介绍
ksqlDB Server 可以将 Kafka topic 映射为一张表，然后执行 SQL 做数据处理，此处不需要使用不过多介绍
Schema Registry 是 AVRO Schema 管理组件， Kafka 中消息 schema 变化时会往 schema registry 发送变更消息，数据同步组件发现 schema 变化之后会从 schema registry 获取最新的 schema 以便同步新 schema 数据到下游数据库
Zookeeper 元数据管理工具，支持 Kafka 选主等
3、配置 TiCDC 任务发生多张表到 Confluent 的 kafka 里的 topic
3.1、在 tiup 服务器里配置 kafka.conf 配置文件
[root@172-16-5-146 kafka]# cat kafka.conf# 指定配置文件中涉及的库名、表名是否为大小写敏感# 该配置会同时影响 filter 和 sink 相关配置，默认为 truecase-sensitive = true# 是否输出 old value，从 v4.0.5 开始支持，从 v5.0 开始默认为 trueenable-old-value = true[filter]# 忽略指定 start_ts 的事务#ignore-txn-start-ts = [1, 2]# 过滤器规则# 过滤规则语法：https://docs.pingcap.com/zh/tidb/stable/table-filter#表库过滤语法rules = ['yz0920.*']
[sink]dispatchers = [{matcher = ['yz0920.*'], topic = "tidb_{schema}_{table}", partition="index-value"},]
复制代码
3.2、创建 ticdc 任务
tiup ctl:v6.1.0 cdc changefeed create --pd="http://172.16.5.146:2399" --sink-uri="kafka://172.16.6.132:9092/tidb-kafka-3?protocol=avro&partition-num=1&max-message-bytes=67108864&replication-factor=1" --changefeed-id="kafka-2" --config="/home/kafka/kafka.conf" --schema-registry="http://172.16.6.132:8081"
[root@172-16-5-146 kafka]# tiup ctl:v6.1.0 cdc changefeed create --pd="http://172.16.5.146:2399" --sink-uri="kafka://172.16.6.132:9092/tidb-kafka-3?protocol=avro&partition-num=1&max-message-bytes=67108864&replication-factor=1" --changefeed-id="kafka-2" --config="/home/kafka/kafka.conf" --schema-registry="http://172.16.6.132:8081"Starting component `ctl`: /root/.tiup/components/ctl/v6.1.0/ctl cdc changefeed create --pd=http://172.16.5.146:2399 --sink-uri=kafka://172.16.6.132:9092/tidb-kafka-3?protocol=avro&partition-num=1&max-message-bytes=67108864&replication-factor=1 --changefeed-id=kafka-2 --config=/home/kafka/kafka.conf --schema-registry=http://172.16.6.132:8081[WARN] This index-value distribution mode does not guarantee row-level orderliness when switching on the old value, so please use caution! dispatch-rules: &config.DispatchRule{Matcher:[]string{"yz0920.*"}, DispatcherRule:"", PartitionRule:"index-value", TopicRule:"tidb_{schema}_{table}"}[2022/09/22 15:14:20.498 +08:00] [WARN] [kafka.go:433] ["topic's `max.message.bytes` less than the `max-message-bytes`,use topic's `max.message.bytes` to initialize the Kafka producer"] [max.message.bytes=1048588] [max-message-bytes=67108864][2022/09/22 15:14:20.498 +08:00] [WARN] [kafka.go:442] ["topic already exist, TiCDC will not create the topic"] [topic=tidb-kafka-3] [detail="{\"NumPartitions\":1,\"ReplicationFactor\":1,\"ReplicaAssignment\":{\"0\":[0]},\"ConfigEntries\":{\"segment.bytes\":\"1073741824\"}}"][2022/09/22 15:14:20.560 +08:00] [WARN] [event_router.go:236] ["This index-value distribution mode does not guarantee row-level orderliness when switching on the old value, so please use caution!"][2022/09/22 15:14:20.561 +08:00] [WARN] [mq_flush_worker.go:98] ["MQ sink flush worker channel closed"][2022/09/22 15:14:20.561 +08:00] [WARN] [mq_flush_worker.go:98] ["MQ sink flush worker channel closed"]Create changefeed successfully!ID: kafka-2Info: {"upstream-id":0,"sink-uri":"kafka://172.16.6.132:9092/tidb-kafka-3?protocol=avro\u0026partition-num=1\u0026max-message-bytes=67108864\u0026replication-factor=1","opts":{},"create-time":"2022-09-22T15:14:20.42496876+08:00","start-ts":436163277049823233,"target-ts":0,"admin-job-type":0,"sort-engine":"unified","sort-dir":"","config":{"case-sensitive":true,"enable-old-value":true,"force-replicate":false,"check-gc-safe-point":true,"filter":{"rules":["yz0920.*"],"ignore-txn-start-ts":null},"mounter":{"worker-num":16},"sink":{"dispatchers":[{"matcher":["yz0920.*"],"dispatcher":"","partition":"index-value","topic":"tidb_{schema}_{table}"}],"protocol":"avro","column-selectors":null,"schema-registry":"http://172.16.6.132:8081"},"cyclic-replication":{"enable":false,"replica-id":0,"filter-replica-ids":null,"id-buckets":0,"sync-ddl":false},"consistent":{"level":"none","max-log-size":64,"flush-interval":1000,"storage":""}},"state":"normal","error":null,"sync-point-enabled":false,"sync-point-interval":600000000000,"creator-version":"v6.1.0"}
复制代码
特别说明：
--sink-uri=“kafka://172.16.6.132:9092” 设置为之前 Confluent 启动的 kafka 地址。
protocol=avro 使用 avro 格式，avro 格式支持 schema 演化，支持自动部分 DDL 同步，对比其它格式有优势
--changefeed-id=“kafka-2” 表示此 ticdc 同步任务 id
--config=“c.conf” 表示 ticdc 任务使用 3.1 中创建的配置文件，实现多表多 topic
--schema-registry=http://172.16.6.132:8081 配合 avro 实现 schema 演化，当 TiDB DDL 变更时会发送消息给 schema-registry ，以便消息组件更新 schema
具体请参考：
https://docs.pingcap.com/zh/tidb/stable/manage-ticdc#sink-uri-%E9%85%8D%E7%BD%AE-kafka
3.3、验证该服务
tiup ctl:v6.1.0 cdc changefeed list –pd=http://172.16.5.146:2399
[root@172-16-5-146 kafka]# tiup ctl:v6.1.0 cdc changefeed list --pd=http://172.16.5.146:2399Starting component `ctl`: /root/.tiup/components/ctl/v6.1.0/ctl cdc changefeed list --pd=http://172.16.5.146:2399[ { "id": "kafka-1", "summary": { "state": "normal", "tso": 436163342428536833, "checkpoint": "2022-09-22 15:18:29.728", "error": null } }, { "id": "kafka-2", "summary": { "state": "normal", "tso": 436163342428536833, "checkpoint": "2022-09-22 15:18:29.728", "error": null } }]
状态 normal 表示数据正常同步
复制代码
备注，如果发现有问题可以删除，命令如下
[root@172-16-5-146 kafka]# tiup ctl:v6.1.0 cdc changefeed remove --pd=http://172.16.5.146:2399 -c kafka-2Starting component `ctl`: /root/.tiup/components/ctl/v6.1.0/ctl cdc changefeed remove --pd=http://172.16.5.146:2399 -c kafka-2
复制代码
4、在 tidb 数据库里添加表并同步到 kafka 里
4.1、在 tidb 数据库里 yz0920 创建表并插入数据
MySQL [(none)]> use yz0920Reading table information for completion of table and column namesYou can turn off this feature to get a quicker startup with -A
Database changed
#创建表CREATE TABLE `lqb0922` ( `id` int(11) NOT NULL, `name` varchar(30) DEFAULT NULL, PRIMARY KEY (`id`) /*T![clustered_index] CLUSTERED */)#插入数据insert into lqb0922 values(1,'sh1');insert into lqb0922 values(2,'sh2');
delete from lqb0922 where id = 1;update lqb0922 set name = 'updated' where id = 2;alter table lqb0922 add column c1 varchar(32);
MySQL [yz0920]> select * from lqb0922 -> ;+----+------+| id | name |+----+------+| 1 | sh1 || 2 | sh2 |+----+------+2 rows in set (0.00 sec)
复制代码
4.2、在 confluent 服务器里查看 kafka 的 topic
kafka-topics --bootstrap-server 127.0.0.1:9092 --lis
复制代码
4.3、重启个消费组进行查看 kafka 里的数据
[root@172-16-6-132 ~]# kafka-console-consumer --bootstrap-server 172.16.6.132:9092 --topic tidb_yz0920_lqb0922 --from-beginningsh1sh2sh3
##返回如下：因为使用 AVRO 格式，所以数据显示并不完全可读，但是可以大致看到上面的数据变更过程sh3是因为中间插入了一条新数据
复制代码
5、使用 Connect 完成 TiDB 实时数据到 Oracle
5.1、安装 connect jdbcSinkConnector 插件
执行以下命令安装 kafka-connect-jdbc 插件，安装之后重启 connect 组件，(可能还需要重启 control center)
confluent-hub install confluentinc/kafka-connect-jdbc:latest
复制代码
5.2、配置 connect 任务
5.2.1. 打开 control connect
http://172.16.6.132:9021/clusters
5.2.2、选择 JdbcSinkConnector 插件
配置好 connect 插件之后登陆 control center ，进入主页面之后点击 connect ，再点击 connect-default , 可以看到之前已经安装的 connect 插件。
5.2.3、配置说明
点击 JdbcSinkConnector 之后输入，配置 Topic ，此处使用正则表达式: tidb_.* 表示匹配所以 TiCDC 任务创建出来的 Kafka Topic
创建 kafka 的 topic
配置 Avro 格式：
配置 Oracle 信息
配置写入选项：
需要特别注意 Insert Mode 需要选择 upsert，它可以支持 update 和 delete，通过报错再次编辑的方式将 Enable deletes 修改为 true，以便支持 delete DML
配置数据转化信息
此处 Table Name Format 配置效果会将 tidb 中 test.t9 写入到 oracle 的 kafka_tidb_test_t9 中国， Primary Key Mode 选择 record_key 以便支持 delete 操作
配置 DDL 支持
按下图配置支持自动创建 Table；自动 DDL 落地：目前验证支持 add column ，不支持 Drop Column，不支持 Rename Column； Quote Identifiers 需要设置为 always 它会在创建表时给表名添加双引号，否则 oracle 会无法创建表
配置 Avro Schema Registry：
需要手工 Add Property，分别输入 key 为 value.converter.schema.registry.url 和 key.converter.schema.registry.url ，然后为 key 添加对应的 Value。
5.2.4、查看配置是否正确
点击 Next ，检查配置是否正确，最终启动任务实现实时同步数据。可以在配置检查页面与以下配置文件进行对比。
{ "name": "JdbcSinkConnectorConnector_2", "config": { "value.converter.schema.registry.url": "http://localhost:8081", "key.converter.schema.registry.url": "http://localhost:8081", "name": "JdbcSinkConnectorConnector_2", "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "tasks.max": "1", "key.converter": "io.confluent.connect.avro.AvroConverter", "value.converter": "io.confluent.connect.avro.AvroConverter", "topics.regex": "tidb_.*", "connection.url": "jdbc:oracle:thin:@172.1.1.1:1521:orcl", "connection.user": "test", "connection.password": "****", "dialect.name": "OracleDatabaseDialect", "connection.attempts": "1", "connection.backoff.ms": "1000", "insert.mode": "upsert", "batch.size": "20", "delete.enabled": "true", "table.name.format": "kafka_${topic}", "pk.mode": "record_key", "auto.create": "true", "auto.evolve": "true", "quote.sql.identifiers": "always", "max.retries": "1", "retry.backoff.ms": "1000" }}
复制代码
5.3、验证 connect DML DDL 实时同步能力
启动任务后可以在页面查看任务运行状态，如果失败则可以通过以下命令查看日志排查问题
滚动查看日志confluent local services connect log -f查看全量日志confluent local services connect log
复制代码
登陆 Oracle 数据库，可以看到 TiDB 中所有表已经自动在 Oracle 中映射创建。
备注，JdbcSinkConnectorConnector_0 截图和配置文件如下：
配置文件 json
{ "name": "JdbcSinkConnectorConnector_0", "config": { "value.converter.schema.registry.url": "http://localhost:8081", "key.converter.schema.registry.url": "http://localhost:8081", "name": "JdbcSinkConnectorConnector_0", "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "tasks.max": "1", "key.converter": "io.confluent.connect.avro.AvroConverter", "value.converter": "io.confluent.connect.avro.AvroConverter", "topics": "tidb_yz0920_lqbyz", "connection.url": "jdbc:oracle:thin:@172.16.11.60:1521:orcl", "connection.user": "ogg", "connection.password": "***", "dialect.name": "OracleDatabaseDialect", "connection.attempts": "5", "connection.backoff.ms": "1000", "insert.mode": "upsert", "batch.size": "20", "table.name.format": "kafka_${topic}", "pk.mode": "record_key", "auto.create": "true", "auto.evolve": "true", "quote.sql.identifiers": "always" }}Launch
复制代码
评论