Flinkcdc 實現 MySQL 寫入 Doris
Flinkcdc 實現 MySQL 寫入 Doris
一、環境配置
Doris：3.0.4 + JDK 17
MySQL （業務數據庫）：5.7
MySQL（本地數據庫）：5.7
Flink：flink-1.19.1
flinkcdc：flink-cdc-3.3.0
二、環境搭建
- 下載 Flink 1.19.1
wget https://archive.apache.org/dist/flink/flink-1.19.1/flink-1.19.1-bin-scala_2.12.tgz
解壓后得到 flink-1.19.1 目錄，設置 FLINK_HOME 為 flink-1.19.1 所在目錄
sudo vim /etc/profile.d/my_env.sh
添加 Flink 路徑
export JAVA_HOME=/opt/module/java
export PATH=$PATH:$JAVA_HOME/bin
export FLINK_HOME=/opt/module/flink-1.19.1
export PATH=$PATH:$FLINK_HOME/bin
- 讀入數據通過在 conf/flink-conf.yaml 配置文件追加下列參數開啟 checkpoint，每隔 3 秒做一次 checkpoint。
execution.checkpointing.interval: 3000
- 使用下面的命令啟動 Flink 集群
./bin/start-cluster.sh
啟動成功的話，可以在 http://localhost:8081/訪問到 Flink Web UI，如下所示：
多次執行 start-cluster.sh 可以拉起多個 TaskManager。
部署 Doris 和 MySQL 這部分省略
添加配置文件
Flink lib 目錄下：
flinkcdc 3.3.0 lib 目錄下
三、通過 Flink CDC CLI 提交任務
編寫任務配置 yaml 文件。 下面給出了一個整庫同步的示例文件 mysql-to-doris.yaml：
################################################################################
# Description: Sync MySQL all tables to Doris
################################################################################
source:type: mysqlhostname: 172.16.11.154port: 3306username: rootpassword: xxxtables: app_db.\.*server-id: 5400-5404server-time-zone: 'Asia/Shanghai'
sink:type: dorisfenodes: 172.16.10.181:8030benodes: 172.16.10.181:8040username: rootpassword: 123456table.create.properties.light_schema_change: truetable.create.properties.replication_num: 1pipeline:name: Sync MySQL Database to Dorisparallelism: 1
其中： source 中的 tables: app_db..* 通過正則匹配同步 app_db 下的所有表。 sink 添加 table.create.properties.replication_num 參數是由于 Docker 鏡像中只有一個 Doris BE 節點。
最后，通過命令行提交任務到 Flink Standalone cluster
bash bin/flink-cdc.sh mysql-to-doris.yaml
提交成功后，返回信息如：
Pipeline has been submitted to cluster.
Job ID: ae30f4580f1918bebf16752d4963dc54
Job Description: Sync MySQL Database to Doris
在 Flink Web UI，可以看到一個名為 Sync MySQL Database to Doris 的任務正在運行。
后續相關信息，參考官網即可
https://nightlies.apache.org/flink/flink-cdc-docs-release-3.3/zh/docs/get-started/quickstart/mysql-to-doris/
的工作原理)
