使用 Flink 消费 MySQL 或 TDSQL-C MySQL 订阅数据

最近更新时间:2023-10-31 15:56:51

我的收藏

操作场景

数据订阅 Kafka 版(当前 Kafka Server 版本为V2.6.0)中,针对 Avro 格式的订阅数据,可以使用 Flink 客户端(仅支持客户端类型为 DataStream API)进行消费,本场景为您提供使用 flink-dts-connector 进行数据消费的 Demo 示例。

前提条件

2. 创建消费组
3. 已安装 Flink 并能够正常执行 Flink 任务。

注意事项

Demo 并不包含消费数据的用法演示,仅对数据做了打印处理,您需要在此基础上自行编写数据处理逻辑,您也可以使用其他语言的 Kafka 客户端消费并解析数据。
目前不支持通过外网连接数据订阅的 Kafka 进行消费,只支持腾讯云内网的访问,并且订阅的数据库实例所属地域与数据消费的地域相同。
DTS 订阅中内置的 Kafka 处理单条消息有一定上限,当源库中的单行数据超过5MB时,订阅任务可能会报错。
在订阅指定库/表对象(非源实例全部),并且采用 Kafka 单分区的场景中,DTS 解析增量数据后,仅将订阅对象的数据写入 Kafka Topic 中,其他非订阅对象的数据会转成空事务写入 Kafka Topic,所以在消费数据时会出现空事务。空事务的 Begin/Commit 消息中保留了事务的 GTID 信息,可以保证 GTID 的连续性和完整性,多个空事务也做了压缩处理以减少消息数量。
为了保证数据可重入,DTS 订阅引入 checkpoint 机制。消息写入 Kafka Server 时,一般每10秒会插入一个checkpoint,用来标识数据同步的位点,在任务中断后再重启识别断点位置,实现断点续传。另外,消费端遇到 checkpoint 消息会做一次 Kafka 消费位点提交,以此来实现消费端数据可重入。

消费 Demo 下载

Demo 中的逻辑讲解及关键参数说明,请参考 Avro Demo 说明(Flink)
Demo 语言
Avro(MySQL/TDSQL-C MySQL)
Java
地址
编译环境:Maven 或者 Gradle 包管理工具,JDK8。用户可自行选择打包工具,如下以 Maven 为例进行介绍。 运行环境:腾讯云服务器(需要与订阅实例相同地域,才能够访问到 Kafka 服务器的内网地址),安装 JRE8。 操作步骤:
1. 下载 Java Flink Demo,然后解压该文件。
2. 进入解压后的目录,为方便使用,目录下分别放置了 Maven 模型文件、pom.xml 文件,用户根据需要选用。 java -jar avro-tools-1.8.2.jar compile -string schema Record.avsc :代码生成路径。
3. 在 pom.xml 文件中修改 Flink 的版本,如下代码中的 version 需要与客户使用的 Flink 版本保持一致。
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
<version>1.13.6</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>1.13.6</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-avro</artifactId>
<version>1.13.6</version>
</dependency>
4. 进入 pom 文件所在的目录,使用 Maven 进行打包,也直接使用 IEDA 打包。 使用 Maven 进行打包:mvn clean package。
5. 针对 Flink 客户端类型为 DataStream API 的场景,使用 Flink 客户端命令提交任务,启动消费。 ./bin/flink run consumerDemo-avro-flink-1.0-SNAPSHOT.jar --brokers xxx --topic xxx --group xxx --user xxx --password xxx —trans2sql
brokers 为数据订阅 Kafka 的内网访问地址,topic 为数据订阅任务的订阅 topic,这两个可在 订阅详情 页查看。
groupuserpassword 分别为消费组的名称、账号和密码,可在 消费管理 页查看。
trans2sql 表示是否转换为 SQL 语句,java 代码中,携带该参数表示转换为 SQL 语句,不携带则不转换。
6. 观察消费情况。
查看正在运行的任务。


查看任务详情。