For eachタスクで大きなパラメーター配列のルックアップテーブルを使用してください。

For each タスクはパラメーター配列をネストされたタスクに反復的に渡し、ネストされた各タスクにその実行に関する情報を提供します。パラメーター配列は5,000文字に制限されており、タスク値参照を使用して渡す場合は48 KBに制限されます。ネストされたタスクに渡すデータ量が多い場合は、入力、タスク値、またはジョブパラメーターを直接使用してそのデータを渡すことはできません。

完全なデータを渡す代替策の1つは、タスクデータをJSONファイルとして保存し、完全なデータの代わりにルックアップキー（JSONデータ内）をタスク入力経由で渡すことです。ネストされたタスクは、キーを使用して各イテレーションに必要な特定のデータを取得できます。

以下の例では、サンプル JSON 構成ファイル、および JSON 構成内の値を検索するネストされたタスクにパラメーターを渡す方法を示します。

サンプル JSON 構成

この構成例はステップのリストであり、各イテレーションにパラメーター（args）があります（この例では3つのステップのみが表示されています）。このJSONファイルは/Workspace/Users/<user>/copy-filtered-table-config.jsonとして保存されていると仮定します。ネストされたタスク内でこれを参照します。

JSON

{
  "steps": [
    {
      "key": "table_1",
      "args": {
        "catalog": "my-catalog",
        "schema": "my-schema",
        "source_table": "raw_data_table_1",
        "destination_table": "filtered_table_1",
        "filter_column": "col_a",
        "filter_value": "value_1"
      }
    },
    {
      "key": "table_2",
      "args": {
        "catalog": "my-catalog",
        "schema": "my-schema",
        "source_table": "raw_data_table_2",
        "destination_table": "filtered_table_2",
        "filter_column": "col_b",
        "filter_value": "value_2"
      }
    },
    {
      "key": "table_3",
      "args": {
        "catalog": "my-catalog",
        "schema": "my-schema",
        "source_table": "raw_data_table_3",
        "destination_table": "filtered_table_3",
        "filter_column": "col_c",
        "filter_value": "value_3"
      }
    }
  ]
}

サンプル For each タスク

ジョブ内のFor eachタスクには、各反復のキーを含む入力が含まれています。copy-filtered-tables この例では、**Inputs** ["table_1","table_2","table_3"]が に設定されている という名前のタスクを示します。このリストは5,000文字に制限されていますが、キーを渡しているだけなので、完全なデータよりもはるかに小さくなります。

この例では、ステップが他のステップやタスクに依存しないため、同時実行数を1より大きく設定して、タスクの実行を高速化できます。

サンプルのネストされたタスク

ネストされたタスクには、親のFor eachタスクから入力が渡されます。この場合、構成ファイル用のKeyとして使用されるように入力を設定します。key次の画像は、値が である**パラメーター**「{{input}} 」を設定しているネストされたタスクを示しています。

このタスクは、コードを含むノートブックです。ノートブックで、以下の Python コードを使用して入力を読み取り、それを構成 JSON ファイルへのキーとして使用できます。JSONファイルからのデータは、テーブルからのデータの読み取り、フィルタリング、書き込みに使用されます。

注記

静的JSONファイルの代わりにライブDeltaテーブルからFor eachタスクを駆動する例については、「コントロールテーブルを使用してFor eachジョブを駆動する」を参照してください。

Python

# copy-filtered-table (iteratable task code to read a table, filter by a value, and write as a new table)

from pyspark.sql.functions import expr
from types import SimpleNamespace

import json

# If the notebook is run outside of a job with a key parameter, this provides
# a default. This allows testing outside of a For each task
dbutils.widgets.text("key", "table_1", "key")

# load configuration (note that the path must be set to valid configuration file)
config_path = "/Workspace/Users/<user>/copy-filtered-table-config.json"
with open(config_path, "r") as file:
    config = json.loads(file.read())

# look up step and arguments
key = dbutils.widgets.get("key")
current_step = next((step for step in config['steps'] if step['key'] == key), None)
if current_step is None:
    raise ValueError(f"Could not find step '{key}' in the configuration")

args = SimpleNamespace(**current_step["args"])

# read the source table defined for the step, and filter it
df = spark.read.table(f"{args.catalog}.{args.schema}.{args.source_table}") \
          .filter(expr(f"{args.filter_column} like '%{args.filter_value}%'"))

# write the filtered table to the destination
df.write.mode("overwrite").saveAsTable(f"{args.catalog}.{args.schema}.{args.destination_table}")