Dinky 上使用 FlinkCDC3.1 PIPELINE 同步MySQL到StarRocks

使用 Dinky 和 FlinkCDC 3.1 实现 MySQL 到 StarRocks 的同步

• 本文介绍如何使用 Dinky 上的 FlinkCDC 3.1 PIPELINE 同步 MySQL 到 StarRocks，实现数据的新增、修改、删除，字段的自动新增和删除同步

注意

• 想用 Flinkcdc Pipeline 到 StarRocks 必须使用 FlinkCDC 3.1，FlinkCDC 3.0的版本会报错显示 Caused by: java.lang.RuntimeException: Cannot find factory with identifier “starrocks” in the classpath.

环境信息

组件版本

Flink 1.17
FlinkCDC 3.1
Dinky 1.1
StarRocks 3.1
Mysql 8.0

Dinky 上使用 FlinkCDC 3.1 Pipeline 同步MySQL到StarRocks需要的依赖

Flink/lib 和 dinky/extends 目录下放置
StarRocks的 Flink connector jar和 MySQL CDC 的 Flink connector jar 以及CDC的jar
mysql-connector-java-8.0.27.jar
flink-sql-connector-mysql-cdc-3.1.0.jar
flink-connector-starrocks-1.2.9_flink-1.17.jar
flink-cdc-pipeline-connector-starrocks-3.1.0.jar
flink-cdc-pipeline-connector-mysql-3.1.0.jar
flink-cdc-dist-3.1.0.jar

注意：

• flink-cdc-dist-3.1.0.jar 这个包需要自己重新编译下 参考这一篇博客 Dinky 使用FLINKCDC 3.1 PIPELINE 同步MySQL到Doris

整库同步配置

FlinkCDC 3.1 Pipeline starrocks会自动建表 默认是主键模型 支持数据删除 修改 字段新增删除

YAML 配置示例

以下是整库同步的配置：

SET 'execution.checkpointing.interval' = '10s';
EXECUTE PIPELINE WITHYAML (
source:type: mysql
  hostname: 152.136.53.49
  port:3306username: root
  password:'xx_12345678'--  tables: test.\.*   整库同步到starrocks的test 库下 除了下游route中的teachers 他会同步到ods.teachers中
  tables: test.\.*
  server-id: 5400-5404sink:type: starrocks
  name: Starrocks Sink
  jdbc-url: jdbc:mysql://127.0.0.1:9030load-url: 127.0.0.1:8030username: root
  password:0123456789table.create.properties.replication_num:1--  table.create.properties.fast_schema_evolution: true starrocks3.2以上支持

route:-source-table: test.teachers
    sink-table: ods.teachers   
    description: route all tables in source_db to sink_db

pipeline:name: Sync MySQL Database to starrocks Route
  parallelism:1
)

说明：

• tables: test..* 表示将 MySQL 数据库 test 中的所有表同步到 StarRocks 的 ods 数据库中。

指定库表同步

• 你也可以选择性地同步特定的表到 StarRocks，并通过 route 配置映射表到指定的目标表。

SET 'execution.checkpointing.interval' = '10s';
EXECUTE PIPELINE WITHYAML (
source:type: mysql
  hostname: 152.136.53.49
  port:3306username: root
  password:'xx-12345'tables: test.teachers,test.teacher -- 指定表同步到StarRocks 配合route sink到指定的库表
  server-id: 5400-5404sink:type: starrocks
  name: Starrocks Sink
  jdbc-url: jdbc:mysql://127.0.0.1:9030load-url: 127.0.0.1:8030username: root
  password:123456table.create.properties.replication_num:1--  table.create.properties.fast_schema_evolution: true   starrocks3.2以上支持

route:-source-table: test.teachers
    sink-table: ods.teachers
  -source-table: test.teacher
    sink-table: ods.teacher
    description: route all tables in source_db to sink_db

pipeline:name: Sync MySQL Database to starrocks Route
  parallelism:1
)

说明：

• 该配置将 MySQL 中的 test.teachers 和 test.teacher 表分别同步到 StarRocks 中的 ods.teachers 和 ods.teacher 表。
• 同样支持数据的新增、修改、删除，以及表结构的演化（字段的新增和删除）。

总结

• 通过 FlinkCDC 3.1，可以轻松实现 MySQL 到 StarRocks 的数据同步，尤其是对于整库或指定表的同步。本文介绍了在 Dinky 上使用 FlinkCDC 3.1 Pipeline 的完整配置，包括自动建表、字段演化、数据的增删改查等操作