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: ソーステーブルに基づいて計算列を追加

# オプション
route:
  # ルーティングルール。ソーステーブルとシンクテーブル間のマッピングを指定します。
  - source-table: flink_test.customers
    sink-table: db.customers_o
    # ルーティングルールを説明するために使用される説明
    description: customers テーブルを同期
  - source-table: flink_test.customers_suffix
    sink-table: db.customers_s
    # ルーティングルールを説明するために使用される説明
    description: customers_suffix テーブルを同期

#オプション
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

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <エンドポイント>
  dbname: <データベース名>
  username: ${secret_values.holousername}
  password: ${secret_values.holopassword}

pipeline:
  name: MySQL データベースを Hologres に同期

(オプション) [ディープチェック] をクリックできます。

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