数据传输服务 数据消费操作指导(TDSQL MySQL 同步至 Kafka)

数据同步到 Kafka 后,可以以通过0.11版本及以上的 Kafka 客户端 进行消费订阅数据,这篇提供了 Java、Go、Python 语言的客户端消费 Demo 示例,方便你快速测试消费数据的流程,了解数据格式解析的方法。

相关产品
数据传输服务 产品页,含版本规格、价格与购买入口。

一、注意事项

1. Demo 并不包含消费数据的用法演示,仅对数据做了打印处理,你需要在此基础上自行编写数据处理逻辑,你也可以使用其他语言的 Kafka 客户端消费并解析数据。

2. 目标 Ckafka 中消息大小设置的上限需要大于源库表中单行数据的最大值,以便数据可以正常同步到目标端。

3. 在同步指定库/表对象(非源实例全部),并且采用 Kafka 单分区的场景中,DTS 解析增量数据后,仅将同步对象的数据写入 Kafka Topic 中,其他非同步对象的数据会转成空事务写入 Kafka Topic,所以在消费数据时会出现空事务。空事务的 Begin/Commit 消息中保留了事务的 GTID 信息,可以保证 GTID 的连续性和完整性。同时,在 MySQL/TDSQL-C MySQL 的消费 Demo 中,多个空事务也做了压缩处理以减少消息数量。

4. 为了保证数据可重入,DTS 同步到 Kafka 链路引入 Checkpoint 机制。消息写入 Kafka Topic 时,一般每10秒会插入一个 Checkpoint,用来标识数据同步的位点,在任务中断后再重启识别断点位置,实现断点续传。另外,消费端遇到 Checkpoint 消息会做一次 Kafka 消费位点提交,以便及时更新消费位点。

5. 数据格式选择 JSON 时,如果你使用过或者熟悉开源订阅工具 Canal,可以选择将这里消费出来的 JSON 格式数据转换成 Canal 工具兼容的数据格式,再进行后续处理,我们的 Demo 中已经提供了相关支持,在启动 Demo 的参数中添加参数 trans2canal 即可实现。目前该功能仅限 Java 语言支持。

二、消费 Demo 下载

·在配置同步任务中,你可以选择不同的数据格式,Avro 和 JSON。Avro 采用二进制格式,消费效率更高,JSON 采用轻量级的文本格式,更加简单易用。选择的数据格式不同,参考的 Demo 示例也不同。

·如下提供的 Demo 示例,均已包含对应的 Avro/JSON 协议文件,无需另外下载。

·Demo 中的逻辑讲解及关键参数说明,请参考 Demo 说明。

Demo 语言 Avro 格式 JSON 格式
Go 地址 地址
Java 地址 地址
Python 地址 地址

三、Java Demo 使用说明

编译环境:Maven 或者 Gradle 包管理工具,JDK8。用户可自行选择打包工具,如下以 Maven 为例进行介绍。
操作步骤:

1. 下载 Java Demo,然后解压该文件。

2. 进入解压后的目录,为方便使用,目录下分别放置了 Maven 模型文件、pom.xml 文件,用户根据需要选用。
使用 Maven 进行打包:mvn clean package。

3. 运行 Demo。
使用 Maven 打包后,进入目标文件夹 target,运行如下代码:java -jar consumerDemo-avro-1.0-SNAPSHOT.jar –brokers xxx –topic xxx –group xxx –trans2sql

·brokers 为 CKafka 的访问地址,topic 为同步任务中设置的 topic 名称,如果是多 topic 需要分别消费。这两个可通过数据同步 > 操作 > 查看获取。

·group为消费组名称,用户可提前在 Ckafka 中创建消费组,也可以在此处自定义输入。

·trans2sql 表示是否转换为 SQL 语句,java 代码中,携带该参数表示转换为 SQL 语句,不携带则不转换。

·trans2canal 表示是否转换为 Canal 格式打印出来,携带该参数表示转换为 Canal 格式,不携带则不转换。

注意:说明:

注意:携带 trans2sql 时,将使用 javax.xml.bind.DatatypeConverter.printHexBinary() 将 byte 值转成16进制,请使用 JDK1.8 版本及以上避免不兼容。如果不需要转 SQL,可以注释此处代码。

4. 观察消费情况。

四、Golang Demo 使用说明

编译环境:Golang 1.12 及以上版本,配置好 Go Module 环境。
操作步骤:

1. 下载 Golang Demo,然后解压该文件。

2. 进入解压后的目录,运行 go build -o subscribe ./main/main.go,生成可执行文件 subscribe。

3. 运行 ./subscribe –brokers=xxx –topic=xxx –group=xxx –trans2sql=true。

·brokers 为 CKafka 的访问地址,topic 为同步任务中设置的 topic 名称,如果是多 topic 需要分别消费。这两个可通过 数据同步>操作>查看 获取。

·group为消费组名称,用户可提前在 Ckafka 中创建消费组,也可以在此处自定义输入。

·trans2sql 表示是否转换为 SQL 语句。

4. 观察消费情况。


五、Python3 Demo 使用说明

编译运行环境:安装 Python3,pip3(用于依赖包安装)。
使用 pip3 安装依赖包:

pip install flagpip install kafka-pythonpip install avro

操作步骤:

1. 下载 Python3 Demo ,然后解压该文件。

2. 运行 python main.py –brokers=xxx –topic=xxx –group=xxx –trans2sql=1。

·brokers 为 CKafka 的访问地址,topic 为同步任务中设置的 topic 名称,如果是多 topic 需要分别消费。这两个可通过数据同步 > 操作 > 查看获取。

·group为消费组名称,用户可提前在 Ckafka 中创建消费组,也可以在此处自定义输入。

·trans2sql 表示是否转换为 SQL 语句。

3. 观察消费情况。

赞 (0)
上一篇 4天前
下一篇 4天前