概述

Dinky 是一个基于 Apache Flink 的实时计算平台，它提供了一站式的 Flink 任务开发、运维、监控等功能。

本教程一步一步的教你如何使用dinky运行CDC pipline任务实现整库同步Starrocks并自动建表功能。Doris同理

前置条件

目前jdk只支持8和11，mysql 5.7以上

已部署好的Dinky
准备好Flink集群

环境准备

FLINK_VERSION : 1.19
dinky-1.19-1.2.0
StarRocks3.3.7
Mysql 8.0.30
Dinky 上使用 FlinkCDC 3.1 Pipeline 同步MySQL到StarRocks需要的依赖

jar包下载目录

Flink/lib 和 dinky/extends 目录下放置
StarRocks的 Flink connector jar和 MySQL CDC 的 Flink connector jar 以及CDC的jar
mysql-connector-java-8.0.30.jar
flink-sql-connector-mysql-cdc-3.2.1.jar
flink-connector-starrocks-1.2.10_flink-1.19.jar
flink-cdc-pipeline-connector-starrocks-3.2.1.jar
flink-cdc-pipeline-connector-mysql-3.2.1.jar
flink-cdc-dist-new.3.2.1.jar
flink-cdc-pipeline-connector-paimon-3.2.1.jar

如果直接在dinky使用flink-cdc-dist-3.2.1.jar 会有java.lang.NoSuchMethodError: org.apache.calcite.tools.FrameworkConfig.getTraitDefs()Lorg/apache/flink/calcite/shaded/com/google/common/collect/ImmutableList;错误，所以我们需要先处理一下

-- 解压 flink-cdc-3.2.1-bin.tar.gz
tar -zxvf flink-cdc-3.2.1-bin.tar.gz       
cd flink-cdc-3.1.0/lib/
-- 解压jar文件·
jar -xvf flink-cdc-dist-3.1.0.jar
--删除冲突包
rm -rf org/apache/calcite
-- 重新打包
jar -cvf  flink-cdc-dist-3.1.0-new.jar * 
把新打包的flink-cdc-dist-3.2.1-new.jar文件放到dinky依赖目录下，重启dinky
所有jar包

开始运行

打开dinky页面，新建Flink Sql任务，输入以下代码,注意把相关IP替换成你自己的

Flink集群需要自己提前注册好，选择对应集群

SET 'execution.checkpointing.interval' = '10s';
EXECUTE PIPELINE WITHYAML (
source:
  type: mysql
  hostname: ip
  port: 3306
  username: root
  password: 'xxx'
  tables: lzs-ad.t_back_data,lzs-ad.t_report
  server-id: 5400-5404
 
sink:
  type: starrocks
  name: Starrocks Sink
  jdbc-url: jdbc:mysql://ip:9030
  load-url: ip:8029
  username: root
  password: 'xxxx'
  table.create.properties.replication_num: 1
  table.create.properties.fast_schema_evolution: true 

route:
  - source-table: lzs-ad.t_back_data
    sink-table: qg-ocpx.t_back_data 
  - source-table: lzs-ad.t_report
    sink-table: qg-ocpx.t_report     

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

运行效果：

错误1:

解决方案：目前调大CN配置参数thrift_rpc_timeout_ms和txn_commit_rpc_timeout_ms超时时间

错误2: