TDSQL Boundless 通过 Binlog 订阅 TDS

本文介绍如何通过 MySQL Binlog 协议直连 TDSQL Boundless 订阅 Binlog。
功能限制
使用 Binlog 订阅 TDSQL Boundless 数据变更时，请注意以下相关限制要求以及暂不支持的场景。
说明：
Binlog 开启与关闭均为平滑进行，无需重启实例，不影响业务读写，通常在5分钟内完成。
仅支持预置资源实例开启或者关闭 Binlog，不支持 Serverless 实例开启或者关闭 Binlog。
Binlog 功能自内核版本 V21.2.0 起正式支持。开启 Binlog 后，暂不支持记录 FUNCTION（函数）、PROCEDURE（存储过程）、TRIGGER（触发器）的 DDL 操作到 Binlog。若存在上述对象的变更，需通过其他方式手动同步至下游。历史内核版本功能限制如下：
内核版本低于 V21.0.0：不支持记录 VIEW（视图）的 DDL 操作到 Binlog。
内核版本低于 V20.0.0：开启 Binlog 后将自动禁用 CREATE TABLE ... AS SELECT ... 语法，如需恢复，需联系技术支持关闭同步功能并重启实例。
不支持 HBase 兼容模式，仅支持 MySQL 兼容模式。
暂不支持开启 冷数据归档 的实例开启 Binlog，且 Binlog 服务与冷数据归档二者只能二选一开启。
暂不支持开启 库表回收站 的实例开启 Binlog，且 Binlog 服务与库表回收站二者只能二选一开启。
暂不支持独立下载 Binlog 文件，可通过 MySQL Binlog Dump 协议或 CKafka 订阅日志，可参考 通过 DTS 订阅 TDSQL Boundless。
Binlog 日志默认可保留 7 天，超出保留期限的 Binlog 将被自动清理，无法恢复。
仅支持运行中的实例开启或关闭 Binlog。
准备工作
1. 按照 管理 Binlog 中开启 Binlog 的步骤，开启实例的 Binlog 日志。
开通成功后，您将获取 Binlog 服务的连接地址，请妥善保管，后续将用于创建数据订阅任务。
2. 创建用于数据订阅的用户，并赋予 REPLICATION CLIENT、REPLICATION SLAVE、PROCESS 和全部对象的 SELECT 权限。
2.1 登录 控制台，在实例列表中，选择已开通 Binlog 的实例 ID。
2.2 进入账号管理页面，单击创建账号，创建用于数据同步的用户。
2.3 在账号所在行，单击操作列修改权限，赋予 REPLICATION CLIENT、REPLICATION SLAVE、PROCESS 和全部对象的 SELECT 权限。
﻿
通过 MySQL 订阅 Binlog
注意：
TDSQL Boundless 仅生成 GTID 模式的 Binlog，建议下游使用 SOURCE_AUTO_POSITION = 1 以获得最佳兼容性。
将下游 MySQL 配置为 TDSQL Boundless 的备机。支持两种拉取方式：GTID 自动定位和指定位点拉取。
下游 MySQL 配置 my.cnf
[mysqld]
server-id = 2
relay-log = mysql-relay-bin
read-only = 1
gtid_mode = ON
enforce_gtid_consistency = ON
方式一：GTID 自动定位（推荐）
在下游的 MySQL 实例上执行如下命令，将自动从 TDSQL Boundless 的 Binlog 服务订阅缺失的事务，无需手动提供日志文件名和位置。
CHANGE REPLICATION SOURCE TO
    SOURCE_HOST = '<binlog_service_vip>',
    SOURCE_PORT = 3307,
    SOURCE_USER = '<subscription_user>',
    SOURCE_PASSWORD = '<password>',
    SOURCE_AUTO_POSITION = 1;
﻿
START REPLICA;
SHOW REPLICA STATUS\\G
方式二：通过指定位点拉取 Binlog
-- 先通过 SHOW BINARY LOGS 和 SHOW BINLOG EVENTS 确定起始位点
CHANGE REPLICATION SOURCE TO
    SOURCE_HOST = '<binlog_service_vip>',
    SOURCE_PORT = 3307,
    SOURCE_USER = '<subscription_user>',
    SOURCE_PASSWORD = '<password>',
    SOURCE_LOG_FILE = '<filename>',
    SOURCE_LOG_POS = '<position>';
﻿
START REPLICA;
通过 Canal 订阅 Binlog
Canal 模拟 MySQL Binlog Dump 协议向 TDSQL Boundless 实例请求 Binlog，解析后投递到下游存储。
配置 instance.properties 文件
# TDSQL Boundless CDC 节点连接信息
canal.instance.master.address = <binlog_service_vip>:3307
canal.instance.dbUsername = <subscription_user>
canal.instance.dbPassword = <password>
canal.instance.connectionCharset = UTF-8
﻿
# 订阅规则（示例：订阅指定库表）
canal.instance.filter.regex = mydb\\..*
﻿
# GTID 模式（推荐）
canal.instance.gtidon = true
﻿
# 起始位点（如需指定位点启动）
# canal.instance.master.position = <filename>:<position>
启动 Canal
# 下载 Canal Deployer
tar -zxvf canal.deployer-*.tar.gz -C /opt/canal
﻿
# 启动
/opt/canal/bin/startup.sh
﻿
# 查看日志
tail -f /opt/canal/logs/canal/canal.log
tail -f /opt/canal/logs/example/example.log
Canal 客户端消费数据
Canal 客户端用法请参见 Canal 官方文档。
通过 Flink CDC 订阅 Binlog
Flink CDC 通过 Debezium 内核模拟 MySQL Binlog Dump 协议，直接连接 Binlog 服务消费 Binlog。
方式一：Flink SQL 方式
-- 创建 CDC Source 表
CREATE TABLE tdsqlb_source (
    id BIGINT,
    name STRING,
    amount DECIMAL(10, 2),
    update_time TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '<binlog_service_vip>',
    'port' = '3307',
    'username' = '<subscription_user>',
    'password' = '<password>',
    'database-name' = 'mydb',
    'table-name' = 'orders',
    'server-time-zone' = 'Asia/Shanghai',
    'scan.incremental.snapshot.enabled' = 'true',
    'scan.startup.mode' = 'latest-offset'  -- 或 'latest-offset' 仅消费增量
);
﻿
-- 写入目标表（如 Kafka、StarRocks、另一个 MySQL 等）
INSERT INTO target_table
SELECT id, name, amount, update_time FROM tdsqlb_source;
方式二：Flink DataStream API 方式（Java）
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
﻿
MySqlSource<String> source = MySqlSource.<String>builder()
    .hostname("<binlog_service_vip>")
    .port(3307)
    .databaseList("mydb")
    .tableList("mydb.orders")
    .username("<subscription_user>")
    .password("<password>")
    .serverTimeZone("Asia/Shanghai")
    .deserializer(new JsonDebeziumDeserializationSchema())
    .build();
﻿
env.fromSource(source, WatermarkStrategy.noWatermarks(), "tdsqlb-cdc-source")
   .print();  // 或写入下游 Sink
方式三：Flink CDC Pipeline 方式（YAML，Flink CDC 3.x）
source:
  type: mysql
  name: TDSQL-B Source
  hostname: <binlog_service_vip>
  port: 3307
  username: <subscription_user>
  password: <password>
  tables: mydb.\\.*
  server-time-zone: Asia/Shanghai
常见问题
详见常见问题 > 数据订阅。
通过 Binlog 订阅 TDSQL Boundless

本页目录：

功能限制

准备工作

通过 MySQL 订阅 Binlog

下游 MySQL 配置 my.cnf

方式一：GTID 自动定位（推荐）

方式二：通过指定位点拉取 Binlog

通过 Canal 订阅 Binlog

配置 instance.properties 文件

启动 Canal

Canal 客户端消费数据

通过 Flink CDC 订阅 Binlog

方式一：Flink SQL 方式

方式二：Flink DataStream API 方式（Java）

方式三：Flink CDC Pipeline 方式（YAML，Flink CDC 3.x）

常见问题
