`For each`タスクを使用して、複数のテーブルを増分でコピーします。

多数のソーステーブルからUnity Catalogテーブルにスケジュールに基づいてデータをコピーする必要がある場合、すべての実行で全行をコピーすると、処理が遅くコストがかかります。各テーブルの最後に処理された行を追跡するには、 ウォーターマーク を使用し、各実行で新しい行のみをコピーします。

このチュートリアルでは、メタデータ駆動型ジョブを構築する方法を示します。

ソーステーブルのリストとそのウォーターマークの状態をDeltaコントロールテーブルに格納します。
For eachタスクを使用して、各テーブルを並列で処理します。
最終成功実行以降に追加された行のみをコピーします
コピーが正常に完了するたびにウォーターマークを更新します。

仕組み

ジョブでは、3つのタスクタイプが順番に連携して使用されます。

タスク	Type	機能
`read_watermarks`	SQL	ウォーターマークコントロールテーブルを読み取り、ソーステーブルごとに1行を返します。
`copy_tables`	For each	`{{tasks.read_watermarks.output.rows}}`を反復処理し、ソーステーブルごとにネストされたタスクを1回実行します
`copy_incremental` (ネスト)	ノートブック	最後のウォーターマーク以降に追加された行を読み取り、それらをターゲットテーブルに書き込み、ウォーターマークを進めます。

タスク	Type	機能
`read_watermarks`	SQL	ウォーターマークコントロールテーブルを読み取り、ソーステーブルごとに1行を返します。
`copy_tables`	For each	`{{tasks.read_watermarks.output.rows}}`を反復処理し、ソーステーブルごとにネストされたタスクを1回実行します
`copy_incremental` (ネスト)	ノートブック	最後のウォーターマーク以降に追加された行を読み取り、それらをターゲットテーブルに書き込み、ウォーターマークを進めます。

SQLタスクの出力 —行オブジェクトのJSON配列— は、For each タスクの**入力**フィールドに{{tasks.read_watermarks.output.rows}}を使用して流れ込みます。ネストされたノートブックは、各イテレーションでsource_table、target_table、watermark_column、およびlast_watermarkを受け取ります。

前提条件

ジョブとノートブックを作成する権限を持つDatabricksワークスペース
Unity Catalogでテーブルを作成する権限
Unity Catalogスキーマ。ここでは、制御テーブルとターゲットテーブルを作成できます (例: config)
SQLタスクを実行するためのSQLウェアハウス
タイムスタンプや整数シーケンスのような単調増加する列を含むソーステーブル

ステップ 1: ウォーターマーク制御テーブルを作成します

ウォーターマークコントロールテーブルは、処理するテーブルを決定し、各テーブルがどれだけコピーされたかを示す信頼できるソースです。各行は1つのソーステーブルを表します。

コントロールテーブルを作成し、2つのソーステーブルを登録するには、次のSQLを実行します。

SQL
CREATE OR REPLACE TABLE config.watermarks (
  source_table      STRING NOT NULL,
  target_table      STRING NOT NULL,
  watermark_column  STRING NOT NULL,
  last_watermark    TIMESTAMP NOT NULL
);

INSERT INTO config.watermarks VALUES
  ('sales.raw_orders',    'sales.orders',    'updated_at', '1970-01-01'),
  ('sales.raw_customers', 'sales.customers', 'updated_at', '1970-01-01');

最初の実行でlast_watermarkを1970-01-01に設定すると、ノートブックは既存のすべての行をコピーし、初期フルロードとして機能します。後続の実行では、前の実行以降に追加または更新された行のみがコピーされます。

ステップ2: コピーノートブックを作成します

ノートブックは、テーブルのイテレーションごとに1回実行されます。ウォーターマークを読み取り、ソースをフィルターし、ターゲットに書き込み、ウォーターマークを進めます。

/Workspace/Users/<username>/copy_incrementalのようなパスにノートブックを作成し、以下のコードを追加します。

Python
# Set defaults for running the notebook outside a job
dbutils.widgets.text("source_table",     "sales.raw_orders",  "Source table")
dbutils.widgets.text("target_table",     "sales.orders",      "Target table")
dbutils.widgets.text("watermark_column", "updated_at",        "Watermark column")
dbutils.widgets.text("last_watermark",   "1970-01-01",        "Last watermark")

source_table     = dbutils.widgets.get("source_table")
target_table     = dbutils.widgets.get("target_table")
watermark_column = dbutils.widgets.get("watermark_column")
last_watermark   = dbutils.widgets.get("last_watermark")

# Read only new rows from the source table
new_rows = spark.table(source_table).filter(
    f"{watermark_column} > '{last_watermark}'"
)

row_count = new_rows.count()
print(f"Copying {row_count} new rows from {source_table}")

if row_count > 0:
    # Append new rows to the target table, creating it if it does not exist
    new_rows.write.format("delta").mode("append").saveAsTable(target_table)

    # Compute the new high-water mark from the rows just written
    from pyspark.sql.functions import max as spark_max
    new_watermark = new_rows.agg(spark_max(watermark_column)).collect()[0][0]

    # Advance the watermark so the next run starts from here
    spark.sql(f"""
        UPDATE config.watermarks
        SET last_watermark = CAST('{new_watermark}' AS TIMESTAMP)
        WHERE source_table = '{source_table}'
    """)

    print(f"Watermark for {source_table} advanced to {new_watermark}")
else:
    print(f"No new rows for {source_table}, watermark unchanged")

dbutils.widgets.text()のデフォルト値を使用すると、ノートブックを直接実行およびテストできます。ノートブックがFor eachタスク内で実行されると、ジョブはこれらのデフォルト値を各イテレーションの実際の値で上書きします。

注記

このノートブックはappendモードを使用しています。このモードは、ソースに挿入のみが含まれている場合に適しています。ソースに更新が含まれている場合は、write.mode("append")の代わりにMERGEステートメントを使用してターゲットテーブルに行をアップサートします。merge 構文については、merge を使用した Delta Lake テーブルへのアップサートを参照してください。

ステップ3: ジョブを作成する

Databricksワークスペースで、サイドバーの ワークフロー をクリックし、次に ジョブ をクリックします。ジョブにIncremental table copyなどの名前を付けます。

ステップ 4: ウォーターマークルックアップタスクを構成します

SQLタスクは制御テーブルを読み込み、その結果をFor eachタスクで利用可能にします。

[タスクを追加] をクリックします。
[タスク名] をread_watermarksに設定します。
Set Type to SQL .

SQL フィールドに、以下を入力します:

SQL
SELECT source_table, target_table, watermark_column, last_watermark
FROM config.watermarks

**SQLウェアハウス**をワークスペース内のウェアハウスに設定します。
「 タスクを作成 」をクリックします。

このタスクが実行されると、Databricks は結果を tasks.read_watermarks.output.rows に JSON 配列としてキャプチャします。

JSON
[
  {
    "source_table": "sales.raw_orders",
    "target_table": "sales.orders",
    "watermark_column": "updated_at",
    "last_watermark": "2024-06-01T12:00:00.000Z"
  },
  {
    "source_table": "sales.raw_customers",
    "target_table": "sales.customers",
    "watermark_column": "updated_at",
    "last_watermark": "2024-06-01T12:00:00.000Z"
  }
]

ステップ 5: `For each`タスクを構成します

For eachタスクは SQL 出力を読み取り、ソーステーブルごとに1つのネストされたタスク実行を起動します。

**タスクを追加**をクリックし、**依存元**をread_watermarksに設定します。
[タスク名] をcopy_tablesに設定します。
Type を For each に設定します。
[入力] フィールドに次を入力します：
```
{{tasks.read_watermarks.output.rows}}
```
[同時実行] を2に設定すると、一度に2つのテーブルをコピーできます。ウェアハウスがより高い並列処理をサポートできる場合は、この値を増やしてください。
ネストされたタスクを構成するには、[ ループするタスクを追加 ] をクリックします。
[タスク名] をcopy_incrementalに設定します。
Set Type to ノートブック .
[パス] をステップ2で作成したノートブックのパスに設定します。
[パラメーター] をクリックし、次に [追加] をクリックして、以下の各パラメーターを追加します。

キー	Value
`source_table`	`{{input.source_table}}`
`target_table`	`{{input.target_table}}`
`watermark_column`	`{{input.watermark_column}}`
`last_watermark`	`{{input.last_watermark}}`

それぞれの{{input.<key>}}参照は、現在のイテレーションの行から対応するフィールドに解決されます。 11. 「 タスクを作成 」をクリックします。

ステップ 6: ジョブを実行し、確認します

今すぐ実行 をクリックしてジョブをトリガーします。
ジョブ実行ページで、copy_tablesノードをクリックしてFor eachタスクを展開します。
実行ページには、イテレーションのテーブルが表示されます。各ソーステーブルにつき1行で、それぞれのステータス、開始時刻、期間が表示されます。
任意のイテレーションをクリックして、ノートブックの出力を表示し、行数とウォーターマークの更新を確認します。

ウォーターマークが進んだことを確認するには、ジョブの完了後に以下のクエリを実行します。

SQL
SELECT source_table, last_watermark FROM config.watermarks;

各last_watermarkの値に、最後にコピーされた行のタイムスタンプが反映されます。値が1970-01-01のままである場合、ソーステーブルにフィルターに一致する行が含まれていなかったか、コピーのタスクでエラーが発生しています。詳細については、タスクの実行出力を確認してください。

パターンを拡張します。

新しいソーステーブルを追加 ：制御テーブルに行を挿入します。次回のジョブ実行で自動的に取得され、1970-01-01からフルロードで開始されます：

SQL
INSERT INTO config.watermarks VALUES
  ('sales.raw_products', 'sales.products', 'updated_at', '1970-01-01');

テーブルを停止する : active列を追加し、SQLタスクでフィルターします。

SQL
ALTER TABLE config.watermarks ADD COLUMN active BOOLEAN DEFAULT TRUE;

-- In the SQL task:
SELECT source_table, target_table, watermark_column, last_watermark
FROM config.watermarks
WHERE active = TRUE

**テーブルをバックフィルします**：そのウォーターマークをリセットし、特定のポイントから再コピーするために

SQL
UPDATE config.watermarks
SET last_watermark = '2024-01-01'
WHERE source_table = 'sales.raw_orders';

その他のリソース

For eachタスクを使用して別のタスクをループで実行します — 同時実行オプションを含むFor eachタスクの設定に関する完全なリファレンス
For eachジョブを駆動するためにコントロールテーブルを使用する — ライブ設定テーブルからFor eachジョブを駆動する
マージを使用してDelta Lakeテーブルにアップサートする — ソースデータに更新が含まれる場合、MERGEを使用して行をアップサートします。
AUTO CDC API：パイプラインでチェンジデータキャプチャをシンプルにします — 挿入、更新、および削除を追跡するソースに対してチェンジデータキャプチャを使用します。

仕組み​

前提条件​

ステップ 1: ウォーターマーク制御テーブルを作成します​

ステップ2: コピーノートブックを作成します​

ステップ3: ジョブを作成する​

ステップ 4: ウォーターマークルックアップタスクを構成します​

ステップ 5: For eachタスクを構成します​

ステップ 6: ジョブを実行し、確認します​

パターンを拡張します。​

その他のリソース​

仕組み