環境準備

通過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表中。

數據類型映射