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

ステップ 1：ノードの作成とタスクの構成

コードレス UI でノードを作成・構成する一般的な手順については、「コードレス UI の構成」ガイドをご参照ください。本トピックでは、これらの手順は繰り返しません。

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

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

データソースおよび送信先を選択した後、リーダーとライター間のカラムマッピングを指定する必要があります。「同名フィールドのマップ」、「同一行のフィールドをマップ」、「マッピングのクリア」、または「マッピングの手動編集」を選択できます。

• Kafka 側には、6 つのデフォルトフィールドがあります。

  フィールド名	説明
  __key__	Kafka レコードのキー。
  __value__	Kafka レコードの値。
  __partition__	Kafka レコードが配置されているパーティション番号。パーティション番号は 0 から始まります。
  __headers__	Kafka レコードのヘッダー。
  __offset__	Kafka レコードのパーティション内におけるオフセット。オフセットは 0 から始まります。
  __timestamp__	Kafka レコードの 13 桁整数ミリ秒タイムスタンプ。

• Kafka 側のフィールドに対してカスタム JSON 解析を構成できます。.（サブフィールドを取得） および []（配列要素を取得） の構文を使用することで、JSON 形式の Kafka レコードの value フィールドから内容を取得できます。

  重要

  JSON フィールド名に「.」文字が含まれる場合、フィールド定義によってそのフィールド値を取得できません。これは、フィールド定義構文において曖昧さが生じるためです。

  以下は、Kafka 内の JSON 形式レコードのデータ値の例です。

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

  • a1 のデータ（"hello"）を同期するには、Kafka 側にフィールド a.a1 を追加します。

  • b のデータ（"world"）を同期するには、Kafka 側にフィールド b を追加します。

  • c のデータ（"yyyyyyy"）を同期するには、Kafka 側にフィールド c[1] を追加します。

  • AA のデータ（"this"）を同期するには、Kafka 側にフィールド d[0].AA を追加します。

  • Kafka 側のフィールドを a.b として定義した場合、データ "unreachable" を同期できません。

• ソースまたは送信先テーブルのフィールドをマッピングから除外できます。マッピングされていないソースフィールドは同期インスタンスによって読み取られず、マッピングされていない送信先フィールドには NULL が書き込まれます。

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

ステップ 4：高度なパラメーターの構成

タスク右側の「高度な構成」をクリックし、「最大予想同時実行数」および「不正データレコードの処理ポリシー」などのパラメーターを設定します。本チュートリアルでは、「不正データレコードの処理ポリシー」を「不正データレコードを無視」に設定し、その他のパラメーターはデフォルト値のままとします。詳細については、「コードレス UI の構成」をご参照ください。

ステップ 5：テストの構成と実行

1. オフライン同期ノードの編集ページ右側で、「Run Configuration」をクリックします。テスト実行用に「リソースグループ」および「スクリプトパラメーター」を設定し、上部ツールバーの「実行」をクリックして、同期リンクが正常に動作するかをテストします。

2. 左側のナビゲーションウィンドウで アイコンをクリックし、次に [個人ディレクトリ] の右側にある アイコンをクリックすると、.sql 拡張子のファイルを作成できます。 次に、以下の SQL クエリを実行して、宛先テーブルのデータが期待どおりであるかを確認できます。

   説明

   • この方法でデータを照会するには、送信先 MaxCompute プロジェクトを DataWorks に計算リソースとしてアタッチ する必要があります。

   • .sql ファイルの編集ページで、右側の「Run Configuration」をクリックします。データソースの「タイプ」、「計算リソース」、および「リソースグループ」を指定し、上部ツールバーの「実行」をクリックします。

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

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

オフライン同期タスク右側の「スケジューリング」をクリックします。「スケジューリング構成」パラメーターを定期実行用に設定し、上部ツールバーの「公開」をクリックして公開パネルを開きます。画面上の指示に従って、タスクを公開 します。

データソースおよび送信先の構成時に、3 つのスケジューリングパラメーター（${startTime}、${endTime}、${partition}）を使用しました。スケジューリング構成では、同期要件に応じて、これらのパラメーターの置き換えポリシーを指定する必要があります。以下に、いくつかの典型的なシナリオの構成例を示します。

目的の間隔に応じてスケジューリングサイクルを設定します。