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

2024/10/23 15:18:23 来源：https://blog.csdn.net/TaiYangb/article/details/142940533 浏览: 次关键词：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: mysqlhostname: 152.136.53.49port: 3306username: rootpassword: 'xx_12345678'--  tables: test.\.*   整库同步到starrocks的test 库下 除了下游route中的teachers 他会同步到ods.teachers中tables: test.\.*server-id: 5400-5404sink:type: starrocksname: Starrocks Sinkjdbc-url: jdbc:mysql://127.0.0.1:9030load-url: 127.0.0.1:8030username: rootpassword: 0123456789table.create.properties.replication_num: 1--  table.create.properties.fast_schema_evolution: true starrocks3.2以上支持route:- source-table: test.teacherssink-table: ods.teachers   description: route all tables in source_db to sink_dbpipeline:name: Sync MySQL Database to starrocks Routeparallelism: 1
)

说明：

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

指定库表同步

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

SET 'execution.checkpointing.interval' = '10s';
EXECUTE PIPELINE WITHYAML (
source:type: mysqlhostname: 152.136.53.49port: 3306username: rootpassword: 'xx-12345'tables: test.teachers,test.teacher -- 指定表同步到StarRocks 配合route sink到指定的库表server-id: 5400-5404sink:type: starrocksname: Starrocks Sinkjdbc-url: jdbc:mysql://127.0.0.1:9030load-url: 127.0.0.1:8030username: rootpassword: 123456table.create.properties.replication_num: 1--  table.create.properties.fast_schema_evolution: true   starrocks3.2以上支持route:- source-table: test.teacherssink-table: ods.teachers- source-table: test.teachersink-table: ods.teacherdescription: route all tables in source_db to sink_dbpipeline:name: Sync MySQL Database to starrocks Routeparallelism: 1
)