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 レコードの値フィールドからコンテンツを取得するには、. (サブフィールドを取得) と [] (配列要素を取得) の構文を使用します。

  重要

  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. オフライン同期ノード編集ページの右側で、[デバッグ設定] をクリックします。テスト実行用に [リソースグループ] と [スクリプトパラメーター] を設定します。次に、上部のツールバーで [実行] をクリックして、同期リンクが正常に実行されるかテストします。

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

   説明

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

   • .sql ファイルの編集ページで、右側の [デバッグ設定] をクリックし、データソースの [タイプ]、[コンピューティングリソース]、および [リソースグループ] を指定します。 次に、上部のツールバーで [実行] をクリックします。

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

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

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

データソースと宛先を設定する際に、${startTime}、${endTime}、${partition} の 3 つのスケジューリングパラメーターを使用しました。スケジューリング設定では、同期のニーズに基づいてこれらのパラメーターの置換ポリシーを指定する必要があります。以下は、いくつかの典型的なシナリオの設定例です。

希望する間隔に基づいてスケジューリング周期を設定します。