DataWorks:単一の Kafka テーブルから MaxCompute へのバッチ同期

ステップ 1: ノードの作成

コードレス UI 構成ガイドに従って、オフライン同期ノードを作成および構成します。このチュートリアルでは、Kafka から MaxCompute への同期に特化した構成の詳細に焦点を当てています。

ステップ 2: データソースと送信先の構成

ステップ 3: フィールドマッピングの構成

ソースと送信先を選択した後、Kafka リーダーと MaxCompute ライターの間で列をマップします。必要に応じて、[同じ名前のフィールドをマップ]、[同一行のフィールドをマップ]、[マッピングのクリア]、または [マッピングの手動編集] を使用します。

デフォルトの Kafka フィールド

Kafka は、MaxCompute カラムに直接マッピングできる 6 つの組み込みフィールドを公開しています。

JSON フィールドの解析

JSON 形式の Kafka 値の場合、カスタムフィールド定義を追加するには、サブフィールドにアクセスするために . を使用し、配列要素にアクセスするために [] を使用します。

このレコード値の例を次に示します。

{
  "a": {
    "a1": "hello"
  },
  "b": "world",
  "c": [
    "xxxxxxx",
    "yyyyyyy"
  ],
  "d": [
    { "AA": "this", "BB": "is_data" },
    { "AA": "that", "BB": "is_also_data" }
  ],
  "a.b": "unreachable"
}

マッピングルール

• マッピングされていないソースフィールドは、同期インスタンスによって読み取られません。

• マッピングされていない送信先フィールドには NULL が書き込まれます。

• 1 つのソースフィールドを複数の送信先フィールドにマッピングすることはできません。

• 1 つの送信先フィールドを複数のソースフィールドからマッピングすることはできません。

ステップ 4: 詳細パラメーターの構成

タスクの右側にある [詳細設定] をクリックします。 このチュートリアルでは、[ダーティデータレコードのポリシー] を [ダーティデータレコードを無視] に設定し、他のすべてのパラメーターはデフォルトのままにします。 パラメーターの詳細については、「コードレス UI 構成」をご参照ください。

ステップ 5: 同期のテスト

1. 編集ページの右側にある[実行設定]をクリックします。テスト実行の[リソースグループ]と[スクリプトパラメータ]を設定し、上部のツールバーにある[実行]をクリックします。

2. テスト実行が完了したら、送信先テーブル内のデータを確認します。左側のナビゲーションウィンドウで、.sql 拡張子のファイルを作成し、次のクエリを実行します。

   説明

   • この方法でデータをクエリするには、送信先 MaxCompute プロジェクトをコンピューティングリソースとして DataWorks にアタッチします。

   • .sql ファイル編集ページで、右側にある [実行構成] をクリックし、[コンピューティングリソース] と [リソースグループ] を指定してから、[実行] をクリックします。

   SELECT * FROM <MaxCompute_destination_table_name> WHERE pt=<specified_partition> LIMIT 20;

ステップ 6: スケジューリングの構成と公開

タスクの右側にある[スケジューリング]をクリックして、定期的な実行のスケジューリング設定を行い、次に[公開]をクリックしてタスクを公開します。

このチュートリアルで使用される 3 つのスケジューリングパラメーター ${startTime}、${endTime}、および ${partition} は互いに関連しています。${startTime} と ${endTime} に注入される値は各インスタンスの Kafka タイムウィンドウを制御し、${partition} はデータを受信する MaxCompute パーティションを決定します。スケジューリングサイクルに基づいて、これら 3 つをまとめて設定してください。

以下の例は、最も一般的なスケジューリングパターンを網羅しています。