利用Flink CDC实现数据同步至Delta Table

环境准备

通过Flink CDC CLI提交任务

1. 下载所需JAR包：

   • flink-cdc包

     进入flink-cdc下载二进制压缩包flink-cdc-3.1.1-bin.tar.gz，并解压得到flink-cdc-3.1.1目录，其中会包含bin、lib、log及conf四个目录，将这四个目录下的文件移动至flink-1.18.0对应的目录下。

   • Connector包

     下载以下Connector包，并移动至flink-1.18.0/lib目录下。

     说明

     下载链接只对已发布的版本有效, SNAPSHOT版本需要本地基于master或release-分支编译。

     • MySQL pipeline connector 3.1.1

     • MaxCompute pipeline connector 3.1.1

   • Driver包

     下载MySQL Connector Java包，通过--jar参数将其传入Flink CDC CLI，或将其放在$flink-1.18.0/lib目录下并重启Flink集群，因为CDC Connectors不再包含这些Drivers。

2. 编写任务配置YAML文件。下述为您提供一个整库同步的示例文件mysql-to-maxcompute.yaml：

   ################################################################################
   # Description: Sync MySQL all tables to MaxCompute
   ################################################################################
   source:
     type: mysql
     hostname: localhost
     port: 3306
     username: root
     password: 123456
     tables: app_db.\.*
     server-id: 5400-5404
     server-time-zone: UTC
   
   # accessId, accessKey, endpoint, project需要用户自行填写
   sink:
      type: maxcompute
      name: MaxComputeSink
      accessId: ${your_accessId}
      accessKey: ${your_accessKey}
      endpoint: ${your_maxcompute_endpoint}
      project: ${your_project}
      bucketsNum: 8
   
   pipeline:
     name: Sync MySQL Database to MaxCompute
     parallelism: 1
   

   参数说明：

   • Source部分的参数配置详情请参见Apache Flink CDC MySQL Connector。

   • Sink部分的参数配置方式请参见连接器Connector配置项。

3. 执行下述命令，提交任务至Flink Standalone集群。

   ./bin/flink-cdc.sh mysql-to-maxcompute.yaml

   提交成功后，返回如下信息：

   Pipeline has been submitted to cluster.
   Job ID: f9f9689866946e25bf151ecc179ef46f
   Job Description: Sync MySQL Database to MaxCompute

   在Flink Web UI中，即可看到一个名为Sync MySQL Database to MaxCompute的任务正在运行。

4. 在MaxCompute中执行如下SQL，查看orders、shipments及products三张表是否已被成功创建，并且可以进行数据写入。

   -- 查看orders表
   read orders;
   
   -- 返回结果：
   +------------+------------+
   | id         | price      |
   +------------+------------+
   | 1          | 4          |
   | 2          | 100        |
   +------------+------------+
   
   -- 查看shipments表
   read shipments;
   
   -- 返回结果
   +------------+------------+
   | id         | city       |
   +------------+------------+
   | 1          | beijing    |
   | 2          | xian       |
   +------------+------------+
   
   -- 查看products表
   read products;
   
   -- 返回结果
   +------------+------------+
   | id         | product    |
   +------------+------------+
   | 3          | Peanut     |
   | 1          | Beer       |
   | 2          | Cap        |
   +------------+------------+

同步变更操作

此处以orders表为例，为您展示在修改MySQL数据库中的源表数据时，MaxCompute中对应的目标表数据也会实时更新。

1. 执行如下命令，进入MySQL容器。

   docker-compose exec mysql mysql -uroot -p123456

2. 在MySQL的orders表中插入一条数据。

   INSERT INTO app_db.orders (id, price) VALUES (3, 100.00);

   在MaxCompute中执行read orders;命令查询orders表数据。返回结果如下：

   +------------+------------+
   | id         | price      |
   +------------+------------+
   | 3          | 100        |
   | 1          | 4          |
   | 2          | 100        |
   +------------+------------+

3. 在MySQL的orders表中增加一个字段。

   ALTER TABLE app_db.orders ADD amount varchar(100) NULL;

   在MaxCompute中执行read orders;命令查询orders表数据。返回结果如下：

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 4          | NULL       |
   | 2          | 100        | NULL       |
   +------------+------------+------------+

4. 在MySQL的orders表中更新一条数据。

   UPDATE app_db.orders SET price=100.00, amount=100.00 WHERE id=1;

   在MaxCompute中执行read orders;命令查询orders表数据。返回结果如下：

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 100        | 100.00     |
   | 2          | 100        | NULL       |
   +------------+------------+------------+

5. 在MySQL的orders表中删除一条数据。

   DELETE FROM app_db.orders WHERE id=2;

   在MaxCompute中执行read orders;命令查询orders表数据。返回结果如下：

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 100        | 100.00     |
   +------------+------------+------------+

对于上述操作，在MySQL中每执行一步，就在MaxCompute中进行一次数据预览，可以看到MaxCompute中显示的orders表数据是实时更新的。

轮询变更操作

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

################################################################################
# Description: Sync MySQL all tables to MaxCompute
################################################################################
source:
   type: mysql
   hostname: localhost
   port: 3306
   username: root
   password: 123456
   tables: app_db.\.*
   server-id: 5400-5404
   server-time-zone: UTC

# accessId, accessKey, endpoint, project 需要用户自行填写
sink:
   type: maxcompute
   name: MaxComputeSink
   accessId: ${your_accessId}
   accessKey: ${your_accessKey}
   endpoint: ${your_maxcompute_endpoint}
   project: ${your_project}
   bucketsNum: 8

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 MaxCompute
   parallelism: 1

route部分的参数详情请参见Flink CDC Route。

通过上面的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中。

环境清理

执行完上述操作后，您需要进行环境清理。

1. 在docker-compose.yml文件所在的目录下执行如下命令停止所有容器：

   docker-compose down

2. 在Flink所在目录flink-1.18.0下，执行如下命令停止Flink集群：

   ./bin/stop-cluster.sh

连接器Connector配置项

表位置映射

连接器Connector自动建表时，使用如下映射关系，将源表的位置信息映射到MaxCompute表中。

数据类型映射