qidao123.com ToB IT社区-企服评测·应用市场»论坛 › 大数据 › 数据仓库与分析 › flink系列之：利用flink cdc3从mysql数据库同步数据到do ...

flink系列之：利用flink cdc3从mysql数据库同步数据到doris和starrocks

[复制链接]
发表于 2025-10-27 03:03:30 | 显示全部楼层 |阅读模式

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

×
一、下载摆设flink



  • 下载flink
解压flink
  1. tar -zxvf flink-1.19.1-bin-scala_2.12.tgz
复制代码
修改flink设置文件config.yaml
  1. taskmanager:
  2.   bind-host: localhost
  3.   host: localhost
  4.   numberOfTaskSlots: 6
  5.   memory:
  6.     process:
  7.       size: 1728m
  9. parallelism:
  10.   default: 1
  11. rest:
  12.   address: 10.66.77.104
  13.   # network interface, such as 0.0.0.0.
  14.   bind-address: 10.66.77.104
  15.   # port: 8081
  16.   # # Port range for the REST and web server to bind to.
  17.   # bind-port: 8080-8090
复制代码
设置flink 环境变零
  1. cd /etc/profile.d
  2. cat flink.sh
  4. #export HADOOP_CLASSPATH=`hadoop classpath`
  5. FLINK_HOME=/data/src/flink/flink-1.19.1
  6. PATH=$PATH:$FLINK_HOME/bin:$FLINK_HOME/sbin
  8. export PATH
  9. export FLINK_HOME
复制代码
启动flink
  1. ./start-cluster.sh
复制代码
查察jps
  1. jps
  2. 760234 StandaloneSessionClusterEntrypoint
  3. 390132 Jps
  4. 760880 TaskManagerRunner
复制代码
查察flink web ui，{ip}:{port}

二、下载摆设flink cdc3



  • https://github.com/apache/flink-cdc/releases

    解压flink-cdc3
  1. tar -zxvf flink-cdc-3.3.0-bin.tar.gz
复制代码
下载Pipeline Connectors Jars和Source Connector Jars到lib目次
  1. /data/src/flink/flink-cdc-3.3.0/lib   ls
  2. flink-cdc-dist-3.3.0.jar                              flink-cdc-pipeline-connector-maxcompute-3.3.0.jar  flink-sql-connector-tidb-cdc-3.3.0.jar
  3. flink-cdc-pipeline-connector-doris-3.3.0.jar          flink-cdc-pipeline-connector-mysql-3.3.0.jar       mysql-connector-java-8.0.28.jar
  4. flink-cdc-pipeline-connector-elasticsearch-3.3.0.jar  flink-cdc-pipeline-connector-paimon-3.3.0.jar
  5. flink-cdc-pipeline-connector-kafka-3.3.0.jar          flink-cdc-pipeline-connector-starrocks-3.3.0.jar
复制代码
三、下载mysql-connector-java到flink和flink cdc的lib目次

https://mvnrepository.com/artifact/mysql/mysql-connector-java/8.0.28

  1. wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.28/mysql-connector-java-8.0.28.jar
复制代码
四、flink设置checkpoint支持增量同步数据



  • execution.checkpointing.interval: 3000
参数分析


  • execution.checkpointing.interval: 这个参数用于指定 Flink 作业实验查抄点的频率。查抄点是 Flink 用于实现容错机制的一种机制，通过定期生存作业的状态，可以在发生故障时规复到近来的一个查抄点。
  • 3000: 这个值体现查抄点的隔断时间，单位是毫秒（ms）。因此，3000 毫秒便是 3 秒。
五、mysql到doris和starrocks的yaml设置文件

放到恣意目次下
mysql-to-doris.yaml
  1.    source:
  2.      type: mysql
  3.      hostname: ip
  4.      port: 3306
  5.      username: *********
  6.      password: ************
  7.      tables: data_entry_test.debeziumOfflineClusterInfo,data_entry_test.debeziumRealtimeClusterInfo
  8.      server-id: 5400-5404
  9.      server-time-zone: Asia/Shanghai
  11.    sink:
  12.      type: doris
  13.      fenodes: ip:8030
  14.      username: ***********
  15.      password: *************
  17.    route:
  18.      - source-table: data_entry_test.debeziumOfflineClusterInfo
  19.        sink-table: optics.debeziumOfflineClusterInfo
  20.      - source-table: data_entry_test.debeziumRealtimeClusterInfo
  21.        sink-table: optics.debeziumRealtimeClusterInfo
  23.    pipeline:
  24.      name: Sync MySQL Database to Doris
  25.      parallelism: 2
复制代码
mysql-to-starrocks.yaml
  1. ################################################################################
  2. # Description: Sync MySQL all tables to Doris
  3. ################################################################################
  4. source:
  5. type: mysql
  6. hostname: ip
  7. port: 3306
  8. username: *********
  9. password: **********
  10. tables: data_entry_test.debeziumOfflineClusterInfo,data_entry_test.debeziumRealtimeClusterInfo
  11. server-id: 5400-5404
  12. server-time-zone: Asia/Shanghai
  14. sink:
  15.   type: starrocks
  16.   name: StarRocks Sink
  17.   jdbc-url: jdbc:mysql://ip:9030
  18.   load-url: ip:8030
  19.   username: ****************
  20.   password: ****************
  21. route:
  22.   - source-table: data_entry_test.debeziumOfflineClusterInfo
  23.     sink-table: dd_test_starrocks.debeziumOfflineClusterInfo
  24.   - source-table: data_entry_test.debeziumRealtimeClusterInfo
  25.     sink-table: dd_test_starrocks.debeziumRealtimeClusterInfo
  26. pipeline:
  27.    name: MySQL to StarRocks Pipeline
  28.    parallelism: 6
复制代码
六、启动flink和flink cdc

启动flink
  1. ./start-cluster.sh
复制代码
启动flink cdc
  1. /data/src/flink/flink-cdc-3.3.0/bin/flink-cdc.sh
  2. /data/src/flink/flink-cdc-3.3.0/conf/mysql-to-starrocks.yaml
复制代码
flink web ui查察使命

七、查察flink cdc使命同步日志日志

  1. 2025-02-18 13:48:49,973 INFO  com.starrocks.connector.flink.catalog.StarRocksCatalog       [] - Success to create table dd_test_starrocks.dd_test_starrocks, sql: CREATE TABLE IF NOT EXISTS dd_test_starrocks.debeziumOfflineClusterInfo (
  2. id VARCHAR(21) NOT NULL,
  3. servername VARCHAR(6168) NOT NULL,
  4. connectorname VARCHAR(6168) NOT NULL,
  5. databasename VARCHAR(6168) NOT NULL,
  6. url VARCHAR(6168) NOT NULL,
  7. topicname VARCHAR(6168) NOT NULL,
  8. clustername VARCHAR(6168) NOT NULL
  9. ) PRIMARY KEY (id)
  10. DISTRIBUTED BY HASH (id);
复制代码
  1. 2025-02-18 14:04:25,298 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Source: Flink CDC Event Source: mysql -> SchemaOperator -> PrePartition (1/2)#0 (2069f3b2a289abd02012736f795a34b7_cbc357ccb763df2852fee8c4fc7d55f2_0_0) switched from INITIALIZING to RUNNING.
  2. 2025-02-18 14:04:25,333 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Source: Flink CDC Event Source: mysql -> SchemaOperator -> PrePartition (2/2)#0 (2069f3b2a289abd02012736f795a34b7_cbc357ccb763df2852fee8c4fc7d55f2_1_0) switched from INITIALIZING to RUNNING.
复制代码
  1. 2025-02-18 14:09:35,729 INFO  com.starrocks.data.load.stream.DefaultStreamLoader           [] - Stream load completed, label : flink-84c2fdac-3341-4b5b-8bf1-3946098c0a97, database : dd_test_starrocks, table : debeziumOfflineClusterInfo, body : {
  2.     "Status": "OK",
  3.     "Message": "",
  4.     "Label": "flink-84c2fdac-3341-4b5b-8bf1-3946098c0a97",
  5.     "TxnId": 108875857,
  6.     "LoadBytes": 133959,
  7.     "StreamLoadPlanTimeMs": 0,
  8.     "ReceivedDataTimeMs": 0
  9. }
复制代码
八、查察mysql表和starrocks表

mysql表
  1. -- data_entry_test.debeziumOfflineClusterInfo definition
  3. CREATE TABLE `debeziumOfflineClusterInfo` (
  4.   `id` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT 'primary key',
  5.   `servername` varchar(2056) NOT NULL COMMENT 'connector标识名',
  6.   `connectorname` varchar(2056) NOT NULL COMMENT 'connector名称',
  7.   `databasename` varchar(2056) NOT NULL COMMENT '数据库名',
  8.   `url` varchar(2056) NOT NULL COMMENT '数据库名',
  9.   `topicname` varchar(2056) NOT NULL COMMENT 'topic名称',
  10.   `clustername` varchar(2056) NOT NULL COMMENT '集群名称',
  11.   `database_server_id` varchar(256) NOT NULL COMMENT '集群名称',
  12.   PRIMARY KEY (`id`)
  13. ) ENGINE=InnoDB AUTO_INCREMENT=765 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
复制代码
starrocks表
  1. -- dd_test_starrocks.debeziumOfflineClusterInfo definition
  3. CREATE TABLE `debeziumOfflineClusterInfo` (
  4.   `id` varchar(21) NOT NULL COMMENT "",
  5.   `servername` varchar(6168) NOT NULL COMMENT "",
  6.   `connectorname` varchar(6168) NOT NULL COMMENT "",
  7.   `databasename` varchar(6168) NOT NULL COMMENT "",
  8.   `url` varchar(6168) NOT NULL COMMENT "",
  9.   `topicname` varchar(6168) NOT NULL COMMENT "",
  10.   `clustername` varchar(6168) NOT NULL COMMENT ""
  11. ) ENGINE=OLAP
  12. PRIMARY KEY(`id`)
  13. DISTRIBUTED BY HASH(`id`)
  14. PROPERTIES (
  15. "replication_num" = "3",
  16. "in_memory" = "false",
  17. "storage_format" = "DEFAULT",
  18. "enable_persistent_index" = "false",
  19. "compression" = "LZ4"
  20. );
复制代码
如上所示，乐成在starrocks表中创建了表，并完成了汗青数据和增量数据的同步
九、flink cdc技能生产环境应用



  • 阿里云基于 Flink CDC 的当代数据栈云上实践
细粒度变动计谋控制：


  • 支持新增表、新增列、修改列名、修改列界说、删除列、删除表和清空表等操纵

当上游数据库新增表时，CDC YAML 可以或许自动辨认并同步这些表的数据，而无需重新设置作业。此功能分为两种环境：


  • 汗青数据同步：通过开启 scan.newly-added-table.enabled 选项，并通过 savepoint 重启作业来读取新增表的汗青数据。
  • 增量数据同步：只需开启 scan.binlog.newly-added-table.enabled 选项，自动同步新增表的增量数据。
回复

使用道具 举报

返回列表

水军大提督
+ 我要发帖

登录后关闭弹窗

登录参与点评抽奖  加入IT实名职场社区
去登录
快速回复 返回顶部 返回列表