MaxCompute:Delta テーブルへのニアリアルタイム書き込みのためのオープンソース Flink CDC。

環境を準備する

Flink CDC CLI を使用して YAML ファイルを送信する

1. 必要な JAR パッケージをダウンロードします。

   • flink-cdc パッケージ

     flink-cdc ページに移動して、バイナリ圧縮パッケージ flink-cdc-3.1.1-bin.tar.gz をダウンロードし、パッケージを解凍して flink-cdc-3.1.1 ディレクトリを取得します。 flink-cdc-3.1.1 ディレクトリには、bin、lib、log、および conf ディレクトリが含まれています。 次に、これらのディレクトリ内のファイルを flink-1.18.0 の関連ディレクトリに移動します。

   • コネクタパッケージ

     次のコネクタパッケージをダウンロードし、flink-1.18.0/lib ディレクトリに移動します。

     説明

     ダウンロードリンクをクリックすると、リリースされたコネクタバージョンのみをダウンロードできます。 SNAPSHOT バージョンを参照する場合は、マスターまたはリリースブランチに基づいて、オンプレミス マシンでバージョンのソースコードをコンパイルする必要があります。

     • MySQL パイプラインコネクタ 3.1.1

     • MaxCompute パイプラインコネクタ 3.1.1

   • ドライバパッケージ

     MySQL Connector/J パッケージをダウンロードし、--jar パラメータを使用して Flink CDC CLI に渡すか、$flink-1.18.0/lib ディレクトリに配置して Flink クラスタを再起動します。 これは、CDC コネクタにこれらのドライバが含まれなくなったためです。

2. タスク構成の YAML ファイルを作成します。 次のサンプルコードは、データベース同期のファイル mysql-to-maxcompute.yaml の例を示しています。

   ################################################################################
   # 説明：MySQL のすべてのテーブルを 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、エンドポイント、およびプロジェクトパラメータを構成します。
   sink:
      type: maxcompute
      name: MaxComputeSink
      accessId: ${your_accessId}
      accessKey: ${your_accessKey}
      endpoint: ${your_maxcompute_endpoint}
      project: ${your_project}
      bucketsNum: 8
   
   pipeline:
     name: MySQL データベースを MaxCompute に同期する
     parallelism: 1
   

   パラメータ：

   • source セクションのパラメータの詳細については、「MySQL コネクタ」をご参照ください。

   • sink セクションのパラメータの詳細については、「コネクタの構成項目」をご参照ください。

3. 次のコマンドを実行して、スタンドアロンモードでデプロイされた Flink クラスタにタスクを送信します。

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

   タスクが送信されると、次の結果が返されます。

   パイプラインはクラスタに送信されました。
   ジョブ ID：f9f9689866946e25bf151ecc179ef46f
   ジョブの説明：MySQL データベースを MaxCompute に同期する

   Flink Web UI では、MySQL データベースを 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 は、ソーステーブルのスキーマまたはデータを他のテーブル名にルーティングする機能を提供します。 この機能を使用して、テーブルまたはデータベース名の置換、データベースの同期などの操作を実行できます。 次のサンプルコードは例を示しています。

################################################################################
# 説明：MySQL のすべてのテーブルを 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、エンドポイント、およびプロジェクトパラメータを構成します。
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: MySQL データベースを MaxCompute に同期する
   parallelism: 1

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

コネクタの構成項目

テーブルの場所のマッピング

Flink CDC コネクタがテーブルを自動的に作成する場合、ソーステーブルの場所情報は、次の表に示すマッピングに基づいて MaxCompute テーブルにマッピングされます。

データ型のマッピング