制御テーブルを使用してFor eachジョブを駆動します

市場、ソーステーブル、顧客、日付パーティションなど、多くの入力に対して同じ処理を実行する場合、そのリストをジョブにハードコーディングすると、リストが変更されるたびにコードを編集して再デプロイする必要があります。代わりに、ジョブが実行時に読み取る コントロールテーブル にリストを保存します。作業を追加または削除するには、テーブル内の行を更新します。次のジョブ実行では、ジョブ自体を編集することなくその変更が反映されます。これは メタデータ駆動型 のパターンです。コードではなくデータが、ジョブが何を処理するかを制御します。

このチュートリアルでは、プリインストールされている Wanderbricks サンプルデータセットでこのパターンを使用するジョブを構築するため、ソースデータを作成することなくエンドツーエンドでランできます。このシナリオは、各 プロパティセグメント (Ski Resort や Urban Year-Round など) に対して同じ価格分析をランするバケーションレンタルプラットフォームです。コントロールテーブルには分析対象のセグメントがリストされ、SQL タスクがそのテーブルを読み取り、For each タスクがセグメントごとに 1 回ずつ、並列で分析をランします。

仕組み

このジョブは、3つのタスクを順番に接続します：

タスク	Type	その機能
read_segments	SQL	コントロールテーブルを読み取り、行をJSON配列としてキャプチャします
process_segments	For each	行配列を反復処理し、ネストされたタスクを行ごとに1回起動します
run_segment_analysis	ノートブックまたはSQL（For each内にネスト）	行ごとに1回ランされ、その行の値を使用して1つのプロパティセグメントを分析します

タスク	Type	その機能
read_segments	SQL	コントロールテーブルを読み取り、行をJSON配列としてキャプチャします
process_segments	For each	行配列を反復処理し、ネストされたタスクを行ごとに1回起動します
run_segment_analysis	ノートブックまたはSQL（For each内にネスト）	行ごとに1回ランされ、その行の値を使用して1つのプロパティセグメントを分析します

フローは read_segments → process_segments → run_segment_analysis です（行ごとに1回）。SQL タスクの出力である行オブジェクトの JSON 配列は、動的値参照 {{tasks.read_segments.output.rows}} を通じて For each タスクの [入力] フィールドに流し込まれます。次に、For each タスクは各行のフィールドをパラメーターとしてネストされたタスクに渡し、{{input.property_type}} および {{input.min_price}} として利用できるようにします。

前提条件

• ジョブとノートブックを作成する権限を持つ Databricks ワークスペース。
• Unity Catalog でテーブルを作成する権限、およびコントロールテーブルを保持するためにカタログ内にスキーマを作成する権限（USE CATALOG および CREATE SCHEMA 権限）。
• SQLタスクを実行するためのSQLウェアハウス。まだ作成していない場合は、「SQLウェアハウスの作成」を参照してください。
• すべての Unity Catalog 有効化ワークスペースで使用可能な samples カタログ。このチュートリアルは samples.wanderbricks.properties から読み取るため、設定が必要なソースデータはありません。

ステップ1：コントロールテーブルを作成する

コントロールテーブルは、ジョブが処理するセグメントリストの信頼できるソースとなります。ジョブの動作を変更するには、ジョブではなくこのテーブルを更新します。

Databricks ノートブックまたは SQL エディタで次の SQL をランします。最初のステートメントはコントロールテーブルを保持するスキーマを作成し、2番目のステートメントはプロパティセグメントごとに1行と、そのセグメントの分析に含める最小リスト価格を持つテーブルを作成します：

SQL

USE CATALOG <catalog-name>;

CREATE SCHEMA IF NOT EXISTS config;

CREATE OR REPLACE TABLE config.property_segments AS
SELECT * FROM VALUES
  ('Urban Year-Round', 150),
  ('Summer Getaway', 200),
  ('Ski Resort', 250)
AS t(property_type, min_price);

<catalog-name> を、ワークスペースカタログなど、スキーマを作成できるカタログに置き換えてください。チュートリアルで config.property_segments を参照しているすべての場所で同じカタログを使用してください。これにはステップ 3 のルックアップクエリーも含まれます。

このステップの後、config.property_segments にはセグメントごとに1行ずつ、合計3行が含まれます。各行には、ジョブが各イテレーションに渡す2つの値（分析対象の property_type と、フィルター対象の min_price の下限値）が含まれます。

ステップ 2: 分析ロジックを記述する

For each タスク内のネストされたタスクは、コントロール テーブルの行ごとに1回実行され、その行の property_type および min_price をパラメーターとして受け取ります。このロジックは、ノートブック タスクまたは SQL タスクとして記述できます。ビジネス ロジックに基づいて選択してください：

• 反復ごとのロジックに手続き型コード、複数言語、またはライブラリが必要な場合（データサイエンスやMachine Learningのステップなど）は、 ノートブックタスク を使用します。
• ロジックが宣言的に表現できる単一のクエリーまたは変換である場合は、 SQLタスク を使用します。SQLタスクには SQLウェアハウスが必要です。

以下の両方のバリアントは同じ結果を生成します。処理対象のセグメントについて、その価格下限以上のリスティング数とその平均価格が生成されます。

Notebook task

/Workspace/Users/<username>/run_segment_analysis などのパスに新しいノートブックを作成します。このノートブックは For each タスクの反復ごとに 1 回実行され、毎回異なるセグメントを受け取ります。

ノートブックに以下のコードを追加してください。

Python

# Set default values so you can run the notebook on its own while developing.
# When the notebook runs inside a For each task, the job overrides these defaults.
dbutils.widgets.text("property_type", "Ski Resort", "Property type")
dbutils.widgets.text("min_price", "250", "Minimum price")

# Read the parameters passed by the For each task.
property_type = dbutils.widgets.get("property_type")
min_price = dbutils.widgets.get("min_price")

result = spark.sql(
    """
    SELECT :property_type AS property_type,
           COUNT(*) AS property_count,
           ROUND(AVG(base_price), 2) AS avg_price
    FROM samples.wanderbricks.properties
    WHERE property_type = :property_type
      AND base_price >= :min_price
    """,
    args={"property_type": property_type, "min_price": min_price},
)
display(result)

注記

dbutils.widgets.get() の前に dbutils.widgets.text() を呼び出してください。最初に get を呼び出すと、ジョブの外部でノートブックを実行した際に InputWidgetNotDefined エラーが発生します。

SQL task

SQLタスクは保存されたクエリーを実行するため、SQLエディタで分析クエリーを作成して保存してください。ステップ4でFor eachタスクを設定する際に、それをネストされたタスクにアタッチします。

1. Databricks ワークスペースで、 [新規] > [クエリー] をクリックして SQL エディタを開きます。

2. 以下のクエリーを入力します。SQLタスクは :param_name 構文を使用してパラメーターを参照するため、クエリーは :property_type および :min_price パラメーターからセグメントと価格の下限を読み取ります：

   SQL

   SELECT :property_type AS property_type,
          COUNT(*) AS property_count,
          ROUND(AVG(base_price), 2) AS avg_price
   FROM samples.wanderbricks.properties
   WHERE property_type = :property_type
     AND base_price >= :min_price;

3. SQLファイルのtab見出しにあるタイトル New Query <date> をクリックし、名前 run_segment_analysis を付けます。次に、 [保存] をクリックして、保存先のフォルダーに移動します。

For each タスクは、実行時に各イテレーションの値を :property_type および :min_price の名前付きパラメーターに渡します。ノートブックウィジェットとは異なり、SQLの名前付きパラメーターはdefault値をサポートしていません。パラメーターが渡されない場合、クエリーはパラメーター解決エラーで失敗します。

ステップ 3: ルックアップクエリーを作成する

ルックアップタスクは、保存されたクエリーを通じてコントロールテーブルを読み取ります。ステップ 2 と同様に、今すぐSQLエディタでクエリーを作成して保存し、ステップ 4 のルックアップタスクにアタッチします。

1. Databricks ワークスペースで、 [新規] > [クエリー] をクリックして SQL エディタを開きます。

2. ステップ 1 で選択したのと同じカタログを使用して、以下を入力します。

   SQL

   SELECT property_type, min_price FROM <catalog-name>.config.property_segments;

   このクエリーを実行するSQL Warehouseが、テーブルを作成したカタログとは異なるカタログをdefaultとして使用する可能性があるため、名前は完全修飾されています。

3. SQLファイルのtab見出しにあるタイトル New Query <date> をクリックし、名前 read_segments を付けます。次に、 [保存] をクリックして、保存先のフォルダーに移動します。

ステップ 4: ジョブを作成および構成する

両方のクエリーを保存したら、ジョブを作成して2つのタスクを追加します。コントロールテーブルを読み取るSQLルックアップタスクと、各行の分析を実行するFor eachタスクです。

ジョブを作成する

Databricks ワークスペースのサイドバーで、 「新規」 > 「ジョブ」 をクリックします。Segment Analysis のような、わかりやすいジョブ名を付けます。

SQL ルックアップタスクを構成する

このタスクはコントロールテーブルを読み取り、ステップ3で保存した read_segments クエリーを実行することで、その行を For each タスクで利用できるようにします。

1. SQLクエリー タイルをクリックして、最初のタスクを設定します。 SQLクエリー タイルが利用できない場合は、 [別のタスクの種類を追加] をクリックし、 SQLクエリー を検索します。
2. タスク名 をread_segmentsに設定します。
3. 必要に応じて、 タイプ ドロップダウン メニューから SQL クエリー を選択します。
4. 「SQLクエリー」 フィールドで、ステップ3で保存した read_segments クエリーを選択します。
5. SQLウェアハウスを ワークスペース内のウェアハウスに設定します。
6. 「 タスクを作成 」をクリックします。

このタスクが実行されると、Databricks は結果を JSON 配列として tasks.read_segments.output.rows にキャプチャします。SQL タスクの出力は常に JSON 配列として返されるため、追加の構成は必要ありません。参照の一般的な形式は tasks.<task-name>.output.rows です。ここで <task-name> は設定したタスク名と一致します。出力は次のようになります。

JSON

[
  { "property_type": "Urban Year-Round", "min_price": 150 },
  { "property_type": "Summer Getaway", "min_price": 200 },
  { "property_type": "Ski Resort", "min_price": 250 }
]

For each タスクを構成する

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

1. [タスクの追加] をクリックし、 [For each] を選択します。

2. タスク名 をprocess_segmentsに設定します。

3. Depends on が read_segments に設定されていることを確認してください。

4. [入力] フィールドに、SQLタスクによってキャプチャされた行配列を入力します。

   
   {{tasks.read_segments.output.rows}}

5. 2つのイテレーションを並列で実行するには、 [同時実行] を 2 に設定します。入れ子のタスクがより高い並列処理をサポートしている場合は、この値を増やしてください。

6. このタスクを完了するには、 [Add a タスク to loop over] をクリックし、各反復で実行されるネストされたタスクを構成します。

For each タスクとそのネストされたタスクは、単一のタスクとして一緒に作成されます。ステップ 2 で選択したタイプに基づいて、ネストされたタスクを構成します。

Notebook task

1. タスク名 をrun_segment_analysisに設定します。

2. Set Type to ノートブック .

3. Path に、ステップ 2 で作成したノートブックを設定します。

4. [パラメーター] をクリックし、 [追加] をクリックして各パラメーターを追加します：

   • キー : property_type 、 値 : {{input.property_type}}
   • キー : min_price 、 値 : {{input.min_price}}

   各 {{input.<key>}} 参照は、現在のイテレーションの行から一致するフィールドに解決されます。

5. [タスクを作成] をクリックして、For eachタスクとそのネストされたタスクをまとめて作成します。

SQL task

このタスクは、ステップ2で保存したrun_segment_analysisクエリーを実行します。

1. タスク名 をrun_segment_analysisに設定します。

2. Type を SQL に設定し、 SQL タスク を クエリー に設定します。

3. [SQLクエリー] フィールドで、ステップ2で保存したrun_segment_analysisクエリーを選択します。

4. SQLウェアハウスを ワークスペース内のウェアハウスに設定します。

5. [パラメーター] をクリックし、 [追加] をクリックして各パラメーターを追加します：

   • キー : property_type 、 値 : {{input.property_type}}
   • キー : min_price 、 値 : {{input.min_price}}

   各 {{input.<key>}} 参照は、現在のイテレーションの行から一致するフィールドに解決されます。

6. [タスクを作成] をクリックして、For eachタスクとそのネストされたタスクをまとめて作成します。

ジョブの有向非巡回グラフ（DAG）に、read_segments から process_segments へのフローが表示され、For each ノード内にネストされたタスクが含まれるようになりました。

ステップ5：ジョブをランして検証する

1. 「今すぐ実行」 をクリックしてジョブを開始してください。
2. ラン tabを選択して、ランを表示します。ジョブの最初のランでは、コンピュートの起動に数分かかります。完了すると、リストに表示されます。
3. process_segmentsノードをクリックして、For eachタスクを展開します。
4. ランページにはイテレーションのテーブルが表示され、セグメントごとに1行ずつ、それぞれのステータス、起動時刻、期間が示されます。
5. いずれかのイテレーション行をクリックして出力を開き、期待されるセグメントが分析されたことを確認します。

各反復の結果を個別に確認できます。特定の反復が失敗した場合、ジョブ全体を再実行することなく、ジョブのランページからその反復のみを再実行できます。

パターンを拡張する

分析にセグメントを追加するには、コントロールテーブルに行を挿入します：

SQL

INSERT INTO <catalog-name>.config.property_segments VALUES ('Historical Place', 100);

次のジョブのランには新しいセグメントが含まれますが、ジョブ構成の変更やノートブックの編集は行われません。

この同じパターンは、データを反復処理の推進力としたいあらゆるケースで機能します：

• 顧客ごとの処理 : 顧客IDごとに1行。ネストされたタスクは、顧客固有の変換を適用するか、顧客固有の宛先に配信します。
• テーブルの取り込み : ソーステーブル名ごとに1行。ネストされたタスクは、各テーブルを読み取って取り込みます。
• バックフィル処理 ：日付パーティションごとに1行。ネストされたタスクは、そのパーティションのヒストリカルデータを再処理します。
• 機能フラグ駆動型実行 : 有効な機能またはエクスペリメントごとに 1 行。ネストされたタスクは、対応するロジックをアクティブ化します。

行を削除せずに処理を停止するには、コントロールテーブルに独自の列（active フラグなど）を追加し、SQLルックアップタスクでその列に基づいてフィルタリングします。これはユーザーが定義して入力する通常の列であり、For each タスクにはこれに関する組み込みの概念はありません。まず列を追加し、次に既存の行を TRUE に設定します。

SQL

ALTER TABLE <catalog-name>.config.property_segments ADD COLUMN active BOOLEAN;
UPDATE <catalog-name>.config.property_segments SET active = TRUE;

次に、アクティブな行のみが反復処理を駆動するように、read_segments クエリーでそれをフィルタリングします：

SQL

SELECT property_type, min_price FROM <catalog-name>.config.property_segments WHERE active = TRUE;

その他のリソース

• For each タスクを使用して別のタスクをループで実行する: For each タスクの構成に関する完全なリファレンス。パラメーターの型と同時実行オプションを含みます。
• 「For eachタスクでの大きなパラメーター配列のルックアップテーブルの使用」: 48 KBのタスク値制限を超える大きなパラメーター配列を処理する方法
• タスクからのパラメーター値へのアクセス：ノートブック、Pythonスクリプト、およびSQLタスクでパラメーター値にアクセスするためのすべてのメソッド
• Wanderbricks データセット: このチュートリアルで使用されるサンプルデータセット