MaxCompute:オープンソース Flink CDC を使用した Delta テーブルへのほぼリアルタイムな書き込み

事前準備

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 の 4 つのディレクトリが含まれています。これらのディレクトリ内のファイルを flink-1.18.0 の対応するディレクトリに移動します。

   • コネクタパッケージ

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

     説明

     ダウンロードリンクは、リリース済みのバージョンにのみ有効です。SNAPSHOT バージョンは、ローカルで master または release- ブランチを基にコンパイルする必要があります。

     • MySQL pipeline connector 3.1.1

     • MaxCompute pipeline connector 3.1.1

   • ドライバーパッケージ

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

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、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 部分のパラメーター設定については、「コネクタの設定項目」をご参照ください。

3. 次のコマンドを実行して、ジョブを Flink スタンドアロンクラスターに送信します。

   ./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 の 3 つのテーブルが正常に作成され、データが書き込まれていることを確認します。

   -- 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 テーブルにデータを 1 件挿入します。

   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 テーブルにフィールドを 1 つ追加します。

   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 テーブルのデータを 1 件更新します。

   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 テーブルからデータを 1 件削除します。

   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、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.yaml ファイルのあるディレクトリで、次のコマンドを実行してすべてのコンテナを停止します。

   docker-compose down

2. Flink のディレクトリ flink-1.18.0 で、次のコマンドを実行して Flink クラスターを停止します。

   ./bin/stop-cluster.sh

コネクタの設定項目

テーブル位置のマッピング

コネクタはテーブルを自動作成する際に、以下のマッピング関係を使用して、ソーステーブルの位置情報を MaxCompute テーブルにマッピングします。

データ型マッピング