TDSQL-subscribe-connector最佳实践（上）

本文介绍了如何在 Oceanus 平台使用 tdsql-subscribe-connector 1 ，从 TDSQL-MySQL 订阅任务 2 创建，到 Oceanus 作业创建、最终数据验证，实现全流程的操作指导。需要注意的是，本文默认已经创建 TDSQL-MySQL 实例和 Oceanus 集群，并且二者在同一 VPC 下或者不同 VPC 下但网络已经打通。

1-tdsql-subscribe-connector数据流程图.png

上述流程图简要说明了使用 tdsql-subscribe-connector 时，整个数据流向情况。TDSQL 的 binlog 数据，会通过订阅任务发送到 Kafka（这里的 Kafka 已经包含在订阅任务中，无需重新创建实例），然后 Oceanus 可以通过 tdsql-subscribe-connector 接入 Kafka 的数据，由于 Kafka 中的消息格式比较特殊，无法用常规 Kafka Connector 接入。

创建订阅任务

创建订阅任务可以参考 数据传输服务 TDSQL MySQL 数据订阅 3 ，在订阅任务创建过程中，需要选择订阅的对象，可以选择不同数据库下的不同表，或者同一数据库下的不同表，当订阅多个表的 binlog 时，多个表中的任意一个的数据变更都会发送到 Kafka ，前提是多个表的 Schema 信息必须是相同的。

例如，以下订阅任务中，就指定了同一个库下的多张表：

2-订阅任务-多表.png

创建 Oceanus SQL 作业

创建 SQL 作业

目前 tdsql-subscribe-connector 仅支持在 SQL 作业中使用，JAR 作业暂时不支持；

在 流计算 Oceanus 控制台 4 的 作业管理 > 新建作业 中新建 SQL 作业，选择在新建的集群中新建作业。然后在作业的 开发调试 > 作业参数 中添加必要的 connector，tdsql-subscribe-connector 目前需要手动上传到依赖管理中，然后在作业参数里引用该 JAR 包，Connector 的 JAR 包文件可以联系腾讯云 Oceanus 团队获取；

3-SQL作业参数.png

创建 Source 端

CREATE TABLE `DataInput` (

      `id` INT,

      `name` VARCHAR

) WITH (

    'connector' = 'tdsql-subscribe',   -- 注意选择对应的内置  Connecto

    'topic' = 'topic-subs-xxx-tdsqlshard-xxx',  -- 替换为订阅任务消费的 Topic

    'scan.startup.mode' = 'latest-offset', -- 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets 的任何一种

    'properties.bootstrap.servers' = 'guangzhou-kafka-2.cdb-dts.tencentcs.com.cn:321',  -- 替换为您的订阅任务 Kafka 连接地址

    'properties.group.id' = 'consumer-grp-subs-xxx-group_2', 

    'format' = 'protobuf', -- 只能是protobuf格式

    'properties.security.protocol'='SASL_PLAINTEXT', -- 认证协议

    'properties.sasl.mechanism'='SCRAM-SHA-512', -- 认证方式

    'properties.sasl.jaas.config'='org.apache.kafka.common.security.scram.ScramLoginModule required username="xxx" password="xxx!";' --用户名和密码

);

正常情况下，以上的 Source 端参数，出了字段定义外，WITH 参数中需要根据具体订阅任务填写；这里列出 Source 端的相关配置项在订阅任务的具体位置：

• topic 数据订阅 > 查看订阅详情 > 订阅信息
• properties.bootstrap.servers 数据订阅 > 查看订阅详情 > 订阅信息
• properties.group.id 数据订阅 > 查看订阅详情 > 消费管理
• properties.sasl.jaas.config 只需要替换 username 和 password 数据订阅 > 查看订阅详情 > 消费管理

创建 Sink 端

-- Logger Sink 可以将输出数据打印到 TaskManager 的日志中  
-- 程序包下载地址：https://github.com/tencentyun/flink-hello-world/releases  
-- 需要先在【程序包管理】中上传该程序包，然后在【作业参数】中引用它  
-- 参见 https://cloud.tencent.com/document/product/849/58713

CREATE TABLE logger_sink_table ( 

    id INT PRIMARY KEY NOT ENFORCED,  

    name STRING 

) WITH ( 

    'connector' = 'logger',

    'print-identifier' = 'DebugData'

);

为了验证方便，这里 Sink 端采用了 Logger 5 ，可以把数据打印到日志文件中，在使用 Logger Connector 前，同样需要下载相关的 JAR ，上传到依赖管理，然后在作业参数中引用；

同时，为了更好地验证日志中数据打印情况，推荐使用 CLS ，可以更方便地在作业控制台查看作业运行日志；

4-作业参数-CLS.png

算子操作

INSERT INTO logger_sink_table SELECT * FROM DataInput;

最后，把 Source 端数据插入到 Sink 端；

结果验证

完成 SQL 作业开发后，发布草稿 > 运行作业 ，然后可以在 Source 表中修改或者新增一些数据:

UPDATE `source_table11` SET `name`='test' WHERE `id`=300001;

INSERT INTO `source_table11` (`id`, `name`) VALUES (6000000, 'test');

DELETE FROM source_table11 WHERE id = 6000000

观察 taskmanager 的日志，可以看到 logger 打印出对应的 RowData 信息：

DebugData-toString: +U(300001,test)

DebugData-toString: +I(6000000,test)

DebugData-toString: -D(6000000,test)

注意事项

1. TDSQL-MySQL 和 Oceanus 的 VPC 需要连通或者使用同一 VPC；
2. 使用 tdsql-subscribe-connector 前，需要构建数据订阅任务；
3. tdsql-subscribe-connector 目前只支持增量阶段，没有全量阶段；
4. 当订阅任务指定了多个表时，多个表的 Schema 需要保持一致；

参考链接

1 tdsql-subscribe-connector: https://cloud.tencent.com/document/product/849/71448

2 订阅任务： https://cloud.tencent.com/document/product/571/68060

3 数据传输服务 TDSQL MySQL 数据订阅: https://cloud.tencent.com/document/product/571/68060

4 流计算 Oceanus 控制台: https://console.cloud.tencent.com/oceanus

5 Logger: https://cloud.tencent.com/document/product/849/58713