Realtime Compute for Apache Flink:Flink CDC データインジェストジョブの開発 (パブリックプレビュー)

対象のワークスペースの[アクション]列にある[コンソール]をクリックします。

左側のナビゲーションウィンドウで、[開発] > [データ取り込み] を選択します。

をクリックして、[テンプレートから作成] をクリックします。

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

現在、MySQL から StarRocks、MySQL から Paimon、および MySQL から Hologres へのテンプレートのみがサポートされています。

ジョブ名、ストレージの場所、およびエンジンバージョンを入力し、[OK] をクリックします。

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

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

対象のワークスペースの[アクション]列にある[コンソール]をクリックします。

左側のナビゲーションウィンドウで、[開発] > [データ取り込み] を選択します。

をクリックし、[新規データ取り込みドラフト] を選択します。[ファイル名] と [エンジンバージョン] を設定し、[作成] をクリックします。

オープンソースコミュニティから Flink CDC ジョブをコピーします。

(任意) [深さチェック] をクリックします。

構文、ネットワーク接続性、アクセス権限を確認できます。

対象のワークスペースの [アクション] 列で、[コンソール] をクリックします。

左側のナビゲーションウィンドウで、[データ開発] > [データ取り込み] を選択します。

をクリックし、[新規データ取り込みドラフト] を選択し、[ファイル名] と [エンジンバージョン] を入力し、[作成] をクリックします。

Flink CDC ジョブを設定します。

# 必須
source:
  # データソースタイプ
  type: <ソースコネクタのタイプに置き換えてください>
  # データソースの設定。設定項目については、対応するコネクタのドキュメントをご参照ください。
  ...

# 必須
sink:
  # 宛先タイプ
  type: <宛先コネクタのタイプに置き換えてください>
  # 宛先の設定。設定項目については、対応するコネクタのドキュメントをご参照ください。
  ...

# オプション
transform:
  # flink_test.customers テーブルの変換ルール
  - 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
    # 変換ルールの説明
    description: append calculated columns based on source table

# オプション
route:
  # ソーステーブルと宛先テーブル間のマッピングを指定するルーティングルール。
  - source-table: flink_test.customers
    sink-table: db.customers_o
    # ルーティングルールの説明
    description: sync customers table
  - source-table: flink_test.customers_suffix
    sink-table: db.customers_s
    # ルーティングルールの説明
    description: sync customers_suffix table

#オプション
pipeline:
  # ジョブ名
  name: MySQL to Hologres Pipeline

コードブロックは次のように説明されます。

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

次のコードは、MySQL の app_db データベースからすべてのテーブルを Hologres のデータベースに同期する方法の例です。

source:
  type: mysql
  hostname: <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: <endpoint>
  dbname: <database-name>
  username: ${secret_values.holousername}
  password: ${secret_values.holopassword}

pipeline:
  name: Sync MySQL Database to Hologres

(任意) [深度チェック] をクリックします。

構文、ネットワーク接続性、アクセス権限を確認できます。