【原创】【ARM适用】Flink的Streaming ELT任务同步MySQL数据到 Doris

da11

【ARM适用】Flink的Streaming ELT任务同步MySQL数据到 Doris

参考帖子：https://nightlies.apache.org/flink/flink-cdc-docs-release-3.3/zh/docs/get-started/quickstart/mysql-to-doris/
Doris部署请查看：https://www.swztdza.cn/dalt/forum.php?mod=viewthread&tid=1201&extra=

警告：部署操作前，请确认mysql或者mariadb数据库版本为5.6版本以上，rpm包安装教程可查看 https://www.swztdza.cn/dalt/forum.php?mod=viewthread&tid=510&highlight=mysql
Mysql官网下载地址：https://downloads.mysql.com/archives/community/
ARM版本只能下载8.0版本以上，所以直接选择8.0版本以上即可。



1. 安装 Flink Standalone 集群
1.1. 下载并解压Flink
点击这里下载 Flink 1.19.1，这里下载到/data/路径下，解压后得到 flink-1.19.1 目录。 使用下面的命令跳转至 Flink 目录下。

cd /data/flink-1.19.1/

1.2. 修改/data/flink-1.19.1/conf/config.yaml 配置文件

在Fault tolerance and checkpointing模块中追加下列参数开启 checkpoint，每隔 3 秒做一次 checkpoint。

execution:
  checkpointing:
    interval: 3min





在Rest & web frontend模块中，修改bind-address: localhost 为 bind-address: 0.0.0.0，这样Web-UI就可以开放外部访问而不用只是限制本机访问了




在Common模块中，找到taskmanager: ，修改numberOfTaskSlots参数为你的cpu核数，这是可用任务的数量，一般建议设置为cpu核数。



保存并退出

1.3. 启动 Flink 集群。

cd /data/flink-1.19.1/

./bin/start-cluster.sh




1.4. 访问Flink-UI
可以在 http://IP:8081/ 访问到 Flink Web UI，如下所示：



多次执行 start-cluster.sh 可以拉起多个 TaskManager。

2. Mysql环境配置
同步任务前，Mysql必须开启Binlog，请在mysql检查以下参数
SHOW VARIABLES LIKE 'log_bin';          -- 必须为 ON
SHOW VARIABLES LIKE 'binlog_format';    -- 必须为 ROW
SHOW VARIABLES LIKE 'binlog_row_image'; -- 必须为 FULL


如果Binlog是关闭状态，请在/etc/my.cnf添加以下参数：

# 启用二进制日志
log_bin = /var/log/mysql/mariadb-bin

# 设置 binlog 格式为 ROW
binlog_format = ROW

# 设置 binlog 行映像为 FULL
binlog_row_image = FULL

# 可选：设置 binlog 过期时间（推荐7天）
expire_logs_days = 7

# 可选：设置 server-id（主从复制必需）
server-id = 1


保存并退出



然后需要重启mysql服务

systemctl restart mysqld

3. 准备Mysql测试数据
进入提前部署好的mysql（注意，必须是mysql5.6版本以上）
-- 创建数据库
CREATE DATABASE app_db;

USE app_db;

-- 创建 orders 表
CREATE TABLE `orders` (
`id` INT NOT NULL,
`price` DECIMAL(10,2) NOT NULL,
PRIMARY KEY (`id`)
);

-- 插入数据
INSERT INTO `orders` (`id`, `price`) VALUES (1, 4.00);
INSERT INTO `orders` (`id`, `price`) VALUES (2, 100.00);

-- 创建 shipments 表
CREATE TABLE `shipments` (
`id` INT NOT NULL,
`city` VARCHAR(255) NOT NULL,
PRIMARY KEY (`id`)
);

-- 插入数据
INSERT INTO `shipments` (`id`, `city`) VALUES (1, 'beijing');
INSERT INTO `shipments` (`id`, `city`) VALUES (2, 'xian');

-- 创建 products 表
CREATE TABLE `products` (
`id` INT NOT NULL,
`product` VARCHAR(255) NOT NULL,
PRIMARY KEY (`id`)
);

-- 插入数据
INSERT INTO `products` (`id`, `product`) VALUES (1, 'Beer');
INSERT INTO `products` (`id`, `product`) VALUES (2, 'Cap');
INSERT INTO `products` (`id`, `product`) VALUES (3, 'Peanut');


4. Doris创建对应的库

Doris 暂时不支持自动创建数据库，需要先创建写入表对应的数据库。

进入 Doris Web UI。
http://ip:8030/
默认的用户名为 admin，默认密码为空。


通过 Web UI 创建 app_db 数据库
create database app_db;



5. 下载Flink CDC及相关包
下载flink-cdc压缩包，并解压得到目录 /data/flink-cdc-3.3.0
flink-cdc-3.3.0 下会包含 bin、lib、log、conf 四个目录。


下载下面列出的 connector 包，并且移动到 /data/flink-cdc-3.3.0/lib 目录下

MySQL pipeline connector 3.3.0
Apache Doris pipeline connector 3.3.0


您还需要将下面的 Driver 包放在 /data/flink-1.19.1/lib/ 目录下，或通过 --jar 参数将其传入 Flink CDC CLI，因为 CDC Connectors 不再包含这些 Drivers：

MySQL Connector Java

6. 编写整库同步任务配置 yaml 文件 mysql-to-doris.yaml：
vim /data/flink-cdc-3.3.0/mysql-to-doris.yaml

################################################################################
## Description: Sync MySQL all tables to Doris
#################################################################################
source:
  type: mysql
  hostname: 127.0.0.1
  port: 3306
  username: root
  password: your_passwd
  tables: app_db.\.*                       # 表示app_db数据库下的所有表都要同步
  server-id: 5400-5404                   # 指定一个范围，确保每个Flink CDC任务中的MySQL源都有唯一的server id，这里设置了一个范围，Flink CDC会自动从中选择一个可用的
  server-time-zone: Asia/Shanghai   # 需要设置和数据库同一个时区，否则会同步失败，这里数据库的时区是Asia/Shanghai（UTF+8）

sink:
  type: doris
  fenodes: 127.0.0.1:8030              # 指定doris地址
  username: root                           # doris用户
  password: ""                               # 对应的密码，默认root和admin是空密码，所以这里留空即可，生产环境强烈建议修改doris密码
  table.create.properties.light_schema_change: true
  table.create.properties.replication_num: 1                            # table.create.properties.replication_num: 1 参数为1是由于只有一个 Doris BE 节点

pipeline:
  name: Sync MySQL Database to Doris
  parallelism: 2


7. 提交任务到 Flink Standalone cluster
cd /data/flink-cdc-3.3.0/
bin/flink-cdc.sh --flink-home /data/flink-1.19.1/ mysql-to-doris.yaml



8. 查看任务
在Flink Web UI，可以看到一个名为 Sync MySQL Database to Doris 的任务正在运行。



如果状态是RUNNING代表执行成功，否则则需要查看日志，



如果需要关闭任务，则点击Cancel Job




9. 查看Doris是否同步
返回Doris-WebUI
点击app_db库，即可看到已经创建出了3个表及同步到对应的数据。



10. 同步变更验证
进入 MySQL，修改 MySQL 数据库中表的数据，Doris 中显示的订单数据也将实时更新：
在 MySQL 的 orders 表中插入一条数据
INSERT INTO app_db.orders (id, price) VALUES (3, 100.00);

在 MySQL 的 orders 表中增加一个字段
ALTER TABLE app_db.orders ADD amount varchar(100) NULL;

在 MySQL 的 orders 表中更新一条数据
UPDATE app_db.orders SET price=100.00, amount=100.00 WHERE id=1;

在 MySQL 的 orders 表中删除一条数据
DELETE FROM app_db.orders WHERE id=2;

每执行一步就刷新一次 Doris Web UI，可以看到 Doris 中显示的 orders 数据将实时更新，如下所示：





同样的，去修改 shipments, products 表，也能在 Doris 中实时看到同步变更的结果。

11. Route the changes
Flink CDC 提供了将源表的表结构/数据路由到其他表名的配置，借助这种能力，我们能够实现表名库名替换，整库同步等功能。 下面提供一个配置文件说明：


################################################################################
# Description: Sync MySQL all tables to Doris
################################################################################
source:
   type: mysql
   hostname: localhost
   port: 3306
   username: root
   password: your_passwd
   tables: app_db.\.*
   server-id: 5400-5404
   server-time-zone: Asia/Shanghai

sink:
   type: doris
   fenodes: 127.0.0.1:8030
   benodes: 127.0.0.1:8040
   username: root
   password: ""
   table.create.properties.light_schema_change: true
   table.create.properties.replication_num: 1

route:
   - source-table: app_db.orders
     sink-table: ods_db.ods_orders
   - source-table: app_db.shipments
     sink-table: ods_db.ods_shipments
   - source-table: app_db.products
     sink-table: ods_db.ods_products

pipeline:
   name: Sync MySQL Database to Doris
   parallelism: 2


通过上面的 route 配置，会将 app_db.orders 表的结构和数据同步到 ods_db.ods_orders 中。从而实现数据库迁移的功能。 特别地，source-table 支持正则表达式匹配多表，从而实现分库分表同步的功能，例如下面的配置：

route:
  - source-table: app_db.order\.*
    sink-table: ods_db.ods_orders


这样，就可以将诸如 app_db.order01、app_db.order02、app_db.order03 的表汇总到 ods_db.ods_orders 中。注意，目前还不支持多表中存在相同主键数据的场景，将在后续版本支持。