Realtime Compute for Apache Flink:Flink CDC データインジェストジョブの開発

対象のワークスペースの Actions 列で、 [Console] をクリックします。

左側メニューで、Development > Data Ingestion を選択します。

をクリックし、 [New Draft with Template] をクリックします。

データ同期テンプレートを選択します。

現在、MySQL から StarRocks、MySQL から Paimon、および MySQL から Hologres へのテンプレートのみが利用可能です。

[name]、 [location]、 [engine version] などのジョブ情報を入力し、 [OK] をクリックします。

Flink CDC ジョブのソースとシンクの情報を設定します。

パラメータ設定の詳細は、関連するコネクタのドキュメントをご参照ください。

対象のワークスペースの [Actions] 列で、 [Console] をクリックします。

左側メニューで、Development > Data Ingestion を選択します。

をクリックし、 [New Draft] を選択します。 [name] と [engine version] を入力し、 [Create] をクリックします。

オープンソースの Flink CDC ジョブの YAML コードをエディターに貼り付けます。

(任意) [Validate] をクリックします。

このオプションは、構文エラー、ネットワーク接続の問題、および権限の問題をチェックします。

対象のワークスペースの [Actions] 列で、 [Console] をクリックします。

左側メニューで、Development > Data Ingestion を選択します。

をクリックし、 [New Draft] を選択します。 [name] と [engine version] を入力し、 [Create] をクリックします。

YAML を使用して Flink CDC ジョブを設定します。例：

# 必須
source:
  # データソースのタイプ。
  type: <ソースコネクタのタイプに置き換えてください>
  # データソースの設定。設定項目の詳細は、対応するコネクタのドキュメントをご参照ください。
  ...
# 必須
sink:
  # シンクのタイプ。
  type: <シンクコネクタのタイプに置き換えてください>
  # シンクの設定。設定項目の詳細は、対応するコネクタのドキュメントをご参照ください。
  ...
# 任意
transform:
  # flink_test.customers テーブルの transform ルール。
  - source-table: flink_test.customers
    # プロジェクション設定。同期するカラムを指定して、データ変換を実行します。
    projection: id, username, UPPER(username) as username1, age, (age + 1) as age1, test_col1, __schema_name__ || '.' || __table_name__ identifier_name
    # フィルター条件。id が 10 より大きいデータのみを同期します。
    filter: id > 10
    # transform ルールの説明。
    description: append calculated columns based on source table
# 任意
route:
  # ソーステーブルとシンクテーブルのマッピングを指定する route ルール。
  - source-table: flink_test.customers
    sink-table: db.customers_o
    # route ルールの説明。
    description: sync customers table
  - source-table: flink_test.customers_suffix
    sink-table: db.customers_s
    # route ルールの説明。
    description: sync customers_suffix table
# 任意
pipeline:
  # ジョブの名前。
  name: MySQL to Hologres Pipeline

次の表に、コードブロックについて説明します。

各モジュールの構文と設定の詳細は、「Flink CDC データインジェストジョブのリファレンス」をご参照ください。

次のコードは、MySQL の app_db データベースのすべてのテーブルを Hologres のデータベースに同期する例を示しています。

source:
  type: mysql
  hostname: <ホスト名>
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: app_db.\.*
  server-id: 5400-5404
  # (任意) 増分フェーズで新しく作成されたテーブルからデータを同期します。
  scan.binlog.newly-added-table.enabled: true
  # (任意) テーブルとフィールドのコメントを同期します。
  include-comments.enabled: true
  # (任意) 無制限のチャンクを優先的にディスパッチして、潜在的な TaskManager の OutOfMemory 問題を防ぎます。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (任意) 読み取りを高速化するため、解析フィルターを有効にします。
  scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
  type: hologres
  name: Hologres Sink
  endpoint: <エンドポイント>
  dbname: <データベース名>
  username: ${secret_values.holousername}
  password: ${secret_values.holopassword}
pipeline:
  name: Sync MySQL Database to Hologres

(任意) [Validate] をクリックします。

このオプションは、構文エラー、ネットワーク接続の問題、および権限の問題をチェックします。