コントロール テーブルを使用して
市場、ソーステーブル、顧客、日付パーティションなど多くの入力で同じ処理を行う場合、そのリストをジョブでハードコーディングすると、リストが変更されるたびにコードを編集し再デプロイすることを意味します。 代わりに、実行時にジョブが読み取る コントロールテーブル にリストを保存します。 作業を追加または削除するには、テーブルの行を更新し、次のジョブ実行でジョブ自体への編集なしで変更を拾います。 これは メタデータ駆動 型のパターンです。コードではなくデータがジョブ処理を制御します。
このチュートリアルでは、このパターンを既存の Wanderbricksサンプルデータセットで使うジョブを構築しており、ソースデータを作成せずにエンドツーエンドで実行できます。 シナリオは、バケーションレンタルプラットフォームが各 不動産セグメント ( Ski Resort や Urban Year-Roundなど)で同じ価格分析を行う場合です。 コントロールテーブルは分析すべきセグメントを一覧にし、SQLタスクはそのテーブルを読み込み、 For each タスクは各セグメントごとに並列に1回ずつ解析を実行します。
どのように機能するのか
この仕事は3つのタスクを連続して結びつけます:
| Task | タイプ | 動作内容 |
|---|---|---|
read_segments |
SQL | コントロールテーブルを読み取り、行をJSON配列としてキャプチャします |
process_segments |
各 | 行配列を反復し、1行ごとにネストタスクを起動します |
run_segment_analysis |
ノートブックまたはSQL( For eachにネストされている) |
各行に1回ずつ実行し、その行の値を使って1つのプロパティセグメントを解析します |
フローは read_segments → process_segments → run_segment_analysis (1列に1回)です。 SQLタスクの出力である行オブジェクトのJSON配列は、動的値参照{{tasks.read_segments.output.rows}}を通じてFor eachタスクのInputsフィールドに流れ込みます。
For eachタスクは各行のフィールドをパラメータとしてネストされたタスクに渡し、{{input.property_type}}と{{input.min_price}}として利用可能です。
前提条件
- ジョブやノートブックを作成する権限を持つAzure Databricksワークスペースです。
- Unity Catalog内でテーブルを作成する権限、そして制御テーブルを保持するための スキーマ (
USE CATALOG権限およびCREATE SCHEMA権限)を作成する権限です。 - SQLタスクを実行するSQLウェアハウス。 もし持っていなければ、 SQL ウェアハウスを作成する方法を参照してください。
-
samplesカタログは、すべてのUnityカタログ対応ワークスペースで利用可能です。 チュートリアルはsamples.wanderbricks.propertiesから読んでいるので、設定すべきソースデータはありません。
ステップ1:コントロールテーブルの作成
コントロールテーブルは、あなたのジョブプロセスのセグメント一覧の真実の情報源です。 ジョブの機能を変更するには、ジョブではなくこのテーブルを更新します。
以下のSQLをAzure DatabricksノートブックまたはSQLエディタで実行してください。 最初の文は制御テーブルを保持するスキーマを作成し、2つ目はプロパティセグメントごとに1行と、そのセグメントの分析に含まれる最低リスティング価格のテーブルを作成します。
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>をワークスペースカタログなど、スキーマを作成できるカタログに置き換えてください。 チュートリアルが参照するすべての場所、ステップ3のルックアップクエリを含め、すべての場所で同じカタログを config.property_segmentsしてください。
このステップの後、 config.property_segments は各セグメントごとに3行ずつ構成されます。 各行には、ジョブが各反復に渡す2つの値、すなわち解析する property_type とフィルタリングする min_price フロアの2つの価値が記されています。
ステップ2:解析ロジックを書く
For eachタスク内のネストされたタスクは、制御テーブルの各行に1回ずつ実行され、その行のproperty_typeとmin_priceをパラメータとして受け取ります。 このロジックはノートタスクやSQLタスクとして書くことができます。 ビジネスの論理に基づいて選択します:
- 反復ごとのロジックで手続き型コード、複数の言語、またはライブラリ（たとえば、データサイエンスや機械学習のステップ）が必要な場合は、ノートブック タスクを使用してください。
- 論理が単一のクエリや変換で宣言的に表現できる場合は SQLタスク を使いましょう。 SQLタスクにはSQLウェアハウスが必要です。
以下の両バリアントは同じ結果を得ます:処理対象セグメントの価格下限以上のリスティング数と平均価格です。
ノートブック タスク
/Workspace/Users/<username>/run_segment_analysisなどのパスに新しいノートブックを作成します。 このノートブックは For each タスクの反復ごとに1回実行され、毎回異なるセグメントを受け取ります。
ノートブックに次のコードを追加します。
# 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)
Note
dbutils.widgets.text()する前にdbutils.widgets.get()を呼び出します。 先に get を呼び出すと、ジョブの外部でノートブックを実行した場合に InputWidgetNotDefined エラーが発生します。
SQL タスク
SQLタスクは保存済みクエリを実行するので、今すぐSQLエディタで解析クエリを作成・保存してください。 ステップ4で For each タスクを設定する際に、ネストされたタスクにアタッチします。
Azure Databricksワークスペースで、
新しい>
SQLエディタを開くためのクエリ。次のクエリを入力します。 SQLタスクは
:param_name構文でパラメータを参照するため、クエリは:property_typeおよび:min_priceパラメータからセグメントおよび価格下限を読み取ります。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;SQLファイルのタブ見出しのタイトル
New Query <date>をクリックし、名前をrun_segment_analysis付けてください。 次に 「保存 」をクリックして、保存したいフォルダに移動します。
For each タスクは、実行時に各反復の値を :property_type および :min_price という名前付きパラメーターに渡します。 ノートブックウィジェットとは異なり、SQL名付きパラメータはデフォルト値をサポートしません。パラメータが渡されなければ、クエリはパラメータ解決エラーで失敗します。
ステップ3:検索クエリの作成
ルックアップタスクは保存されたクエリを通じて制御テーブルを読み込みます。 ステップ2と同様に、今すぐSQLエディタでクエリを作成・保存し、ステップ4のルックアップタスクにアタッチします。
Azure Databricksワークスペースで、
新しい>SQLエディタを開くためのクエリ。ステップ 1 で選択したものと同じカタログを使用して、以下を入力してください:
SELECT property_type, min_price FROM <catalog-name>.config.property_segments;この名前は完全に限定されているのは、このクエリを実行するSQLウェアハウスが、あなたがテーブルを作成したものとは異なるカタログをデフォルトで使う可能性があるからです。
SQLファイルのタブ見出しのタイトル
New Query <date>をクリックし、名前をread_segments付けてください。 次に 「保存 」をクリックして、保存したいフォルダに移動します。
ステップ4:ジョブの作成と設定
両方のクエリを保存した状態でジョブを作成し、その2つのタスクを追加します。制御テーブルを読み取るSQLルックアップタスクと、各行の解析を実行する For each タスクです。
仕事を創出しましょう
Azure Databricksのワークスペースで、サイドバーの新しい>
仕事。 仕事に説明的な名前を付け、例えば Segment Analysis。
SQLルックアップタスクの設定
このタスクはコントロールテーブルを読み取り、ステップ3で保存したread_segmentsクエリを実行することで、その行をFor eachタスクに提供します。
- SQL クエリ タイルをクリックして最初のタスクを設定してください。 SQL クエリ タイルが利用できない場合は、「 別のタスクタイプを追加 」をクリックして SQLクエリを検索してください。
-
タスク名を
read_segmentsに設定します。 - 必要に応じて、タイプドロップダウンメニューからSQLクエリを選択します。
-
SQLクエリ欄で、ステップ3で保存した
read_segmentsクエリを選択します。 - SQL ウェアハウスをワークスペース内のウェアハウスに設定します。
- [ タスクの作成] をクリックします。
このタスクを実行すると、Azure Databricksは結果をtasks.read_segments.output.rowsのJSON配列としてキャプチャします。 SQLタスクの出力は常にJSON配列として返されるので、追加の設定は不要です。 参照の一般的な形は tasks.<task-name>.output.rowsで、 <task-name> は設定したタスク名に一致します。 出力は次のようになります。
[
{ "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 つの入れ子になったタスク実行を起動します。
-
タスクを追加し、「それぞれについて」を選択します。
タスク名を
process_segmentsに設定します。Depends onが
read_segmentsに設定されているか確認してください。入力フィールドに、SQLタスクでキャプチャされた行配列を入力します:
{{tasks.read_segments.output.rows}}2 つの反復を並列に実行するには、Concurrency を
2に設定します。 入れ子になったタスクが並列処理の向上をサポートしている場合は、この値を大きくします。このタスクを完了するには、「 タスクを追加 」をクリックしてループし、各反復で実行されるネストされたタスクを設定してください。
For eachタスクとその入れ子タスクは、単一のタスクとして一緒に作成されます。 ステップ2で選択したタイプに基づいてネストタスクを設定しましょう:
ノートブック タスク
タスク名を
run_segment_analysisに設定します。種類 を ノートブック に設定します。
Step 2で作成したノートに パス を設定します。
パラメータをクリックし、追加をクリックして各パラメータを追加してください:
-
キー:
property_type、 値:{{input.property_type}} -
キー:
min_price、 値:{{input.min_price}}
各
{{input.<key>}}参照は、現在の反復処理対象の行にある対応するフィールドに解決されます。-
キー:
「 タスクを作成 」をクリックすると、
For eachタスクとその入れ子タスクを一緒に作成できます。
SQL タスク
このタスクは、Step 2で保存した run_segment_analysis クエリを実行します。
タスク名を
run_segment_analysisに設定します。TypeをSQLに設定し、SQLタスクをQueryに設定します。
SQLクエリフィールドで、ステップ2で保存した
run_segment_analysisクエリを選択します。SQL ウェアハウスをワークスペース内のウェアハウスに設定します。
パラメータをクリックし、追加をクリックして各パラメータを追加してください:
-
キー:
property_type、 値:{{input.property_type}} -
キー:
min_price、 値:{{input.min_price}}
各
{{input.<key>}}参照は、現在の反復処理対象の行にある対応するフィールドに解決されます。-
キー:
「 タスクを作成 」をクリックすると、
For eachタスクとその入れ子タスクを一緒に作成できます。
ジョブの有向非巡回グラフ (DAG) には、read_segments から process_segments への処理の流れが表示され、入れ子になったタスクは For each ノード内に表示されます。
ステップ5:作業を実行して確認する
- [ 今すぐ実行 ] をクリックしてジョブをトリガーします。
- ランニングタブを選択してランを見てください。 ジョブの最初の実行は計算開始に数分かかります。完成するとリストに表示されます。
-
process_segmentsノードをクリックするとFor eachタスクを展開できます。 - 実行ページには、各セグメントに1行ずつの反復のテーブルがあり、それぞれの状態、開始時間、持続時間が記載されています。
- 任意の反復行をクリックすると出力を開き、期待されるセグメントが解析されたか確認できます。
各反復の結果は独立して確認できます。 特定の反復が失敗した場合は、ジョブ実行ページからその反復だけを実行でき、ジョブ全体をやり直す必要はありません。
パターンを拡張する
分析にセグメントを追加するには、制御表に行を挿入します:
INSERT INTO <catalog-name>.config.property_segments VALUES ('Historical Place', 100);
次のジョブ実行では新しいセグメントが含まれ、ジョブ設定の変更やノートブックの編集はありません。
この同じパターンは、データを基に反復処理を行いたいあらゆるケースで使えます:
- 顧客ごとの処理: 顧客 ID ごとに 1 行。 ネストタスクは顧客固有の変換を適用したり、顧客固有の目的地に納品します。
- テーブル インジェスト: ソース テーブル名ごとに 1 行。 ネストされたタスクは各テーブルを読み取り、取り込みます。
- バックフィル処理: 日付パーティションごとに 1 行。 ネストされたタスクは、そのパーティションの過去データを再処理します。
- 機能フラグドリブン実行: 有効な機能または実験ごとに 1 行。 ネストされたタスクは対応するロジックを起動します。
削除せずに処理を停止するには、自分のカラム( active フラグなど)をコントロールテーブルに追加し、SQLルックアップタスクでフィルタリングしてください。 これは定義して入力する通常の列です。 For each タスクには組み込みの概念はありません。 まず列を追加し、既存の行を TRUEに設定します。
ALTER TABLE <catalog-name>.config.property_segments ADD COLUMN active BOOLEAN;
UPDATE <catalog-name>.config.property_segments SET active = TRUE;
その後、 read_segments クエリでその点をフィルターし、アクティブな行だけが反復を駆動するようにします:
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データセット:このチュートリアルで使用されたサンプルデータセット
