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

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

左側のナビゲーションウィンドウで、[開発] > [データインジェスト] を選択します。

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

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

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

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

Flink CDC ジョブのソースと sink の情報を構成します。

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

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

左側のナビゲーションウィンドウで、[開発] > [データインジェスト] を選択します。

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

オープンソースの Flink CDC ジョブのコードをコピーします。

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

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

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

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

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

Flink CDC ジョブを構成します。

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

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

# オプション
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:
  # ルーティングルール。ソーステーブルと sink テーブルのマッピングを指定します。
  - 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

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

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