タスク値 は、Databricks Utilities taskValues サブユーティリティを参照します。これにより、Databricks ジョブ内のタスク間で任意の値を渡すことができます。 「taskValues サブユーティリティ (dbutils.jobs.taskValues)」を参照してください。
1 つのタスクで dbutils.jobs.taskValues.set() を使用してキーと値のペアを指定し、タスク名とキーを使用して後続のタスクの値を参照できます。
注
dbutils.jobs.taskValues.set() サブユーティリティの dbutils.jobs.taskValues.get() と dbutils.jobs.taskValues は Python 関数であるため、Python が言語として選択されているノートブックでのみ使用できます。 ただし、パラメーターをサポートするすべてのタスクの動的値参照を使用して、タスク値を参照できます。 「タスク値を参照する」を参照してください。
タスク値を設定する
dbutils.jobs.taskValues.set()を使用して Python ノートブックでタスク値を設定します。
タスク値キーは文字列である必要があります。 ノートブックで複数のタスク値が定義されている場合、各キーは一意である必要があります。
手動またはプログラムでタスク値をキーに割り当てることができます。 有効な JSON として表現できる値のみが許可されます。 値の JSON 表現のサイズは 48 KiB 以下にする必要があります。
たとえば、次の例では、キー fave_food の静的文字列を設定します。
dbutils.jobs.taskValues.set(key = "fave_food", value = "beans")
以下の例は、ノートブックタスクパラメータを使って、特定の予約のすべての更新レコードをクエリし、現在の予約状況とレコード総数を返します。
from pyspark.sql.functions import col
dbutils.widgets.text("booking_id", "51567", "Booking ID")
booking_id = dbutils.widgets.get("booking_id")
query = (spark.read.table("samples.wanderbricks.booking_updates")
.orderBy(col("updated_at"), ascending=False)
.where(col("booking_id") == booking_id)
.select(col("status"))
)
dbutils.jobs.taskValues.set(key = "record_count", value = query.count())
dbutils.jobs.taskValues.set(key = "booking_status", value = query.take(1)[0][0])
このパターンを使って値のリストを渡し、それを For each タスクなどの下流ロジックの調整に使うことができます。 「For each タスクを使用して別のタスクをループで実行する」を参照してください。
以下の例は、Pythonリストに宛先IDの異なる値を抽出し、これをタスク値として設定しています:
dest_list = list(spark.read.table("samples.wanderbricks.properties").select("destination_id").distinct().toPandas()["destination_id"])
dbutils.jobs.taskValues.set(key = "dest_list", value = dest_list)
タスク値を参照する
Databricks では、動的値参照パターン {{tasks.<task_name>.values.<value_name>}} を使用して構成されたタスク パラメーターとしてタスク値を参照することをお勧めします。
たとえば、dest_list というタスクのキー destination_lookup を使用してタスク値を参照するには、構文 {{tasks.destination_lookup.values.dest_list}} を使用します。
「タスク パラメーターの構成」と「動的な値の参照」を参照してください。
dbutils.jobs.taskValues.get を使用します
構文 dbutils.jobs.taskValues.get() では、アップストリーム タスク名を指定する必要があります。 複数のダウンストリーム タスクでタスク値を使用できるため、この構文はお勧めしません。つまり、タスク名が変更された場合は多数の更新が必要になります。
この構文を使用して、必要に応じて default 値と debugValue を指定できます。 既定値は、キーが見つからない場合に使用されます。
debugValue を使用すると、ノートブックをタスクとしてスケジュールする前に、ノートブックで手動でコードを開発し、テストするときに使用する静的な値を設定できます。
次の例では、タスク名 booking_status に設定されたキー booking_lookup の値を取得します。 値 confirmed は、ノートブックを対話的に実行する場合にのみ返されます。
booking_status = dbutils.jobs.taskValues.get(taskKey = "booking_lookup", key = "booking_status", debugValue = "confirmed")
注
Databricks では、既定値の設定は推奨されません。キーの欠如やタスクの名前が正しくないため、トラブルシューティングが難しく、予期されるエラー メッセージを防ぐのが困難な場合があるためです。
For eachタスクの反復から値を読み取る
「 各タスクについて 」内で実行されるタスクは、1回の反復に1回実行され、各反復で dbutils.jobs.taskValues.set() でタスク値を設定できます。 下流タスクは、反復インデックス順に並んだ単一の返還リストから各反復の値を読み取ることができます。
Important
For eachタスクの集約された反復出力の閲覧はベータ版機能です。 デフォルトで有効化されており、クラシックコンピュートではDatabricks Runtime 15.4 LTS以上が必要です。
サーバーレス コンピューティングでは、dbutils.jobs.taskValues.get以上が必要です。 環境バージョン 1 から 4 では、リストではなく最初の反復の値のみが返され、TypeError が発生する可能性があります。 動的値参照は任意の計算で完全なリストを返します。
どちらの読み取りサーフェスでも、For each タスク内で実行されるネストされたタスクのキーを使用し、For each タスク自体のキーは使用しないでください。 キーは各反復の値を、反復インデックス順に並べた単一のリストとして返します。 たとえば、process という名前のネストされたタスクの 3 回の反復それぞれが result を設定する場合:
results = dbutils.jobs.taskValues.get(taskKey = "process", key = "result")
# results == [1, 2, 3]
動的価値参照 {{tasks.process.values.result}} は同じ配列に解決されます。
キーを設定しなかった反復は結果内での位置を保持し、dbutils.jobs.taskValues.getからはNone(JSON null)として現れ、動的価値参照ではnullとして現れます。 例えば、3回目の反復のうち2回目が resultを設定しなかった場合、 results は [1, None, 3]です。
集約された参照には2つの制限があります。アセンブル配列は単一のパラメータ値に対して約48 KB(49,344文字)を超えてはならず、また単一のパラメータ値には最大3つの集約されたタスク値参照を含めることができます。
タスク値を表示する
各実行のタスク値の戻り値が、タスク実行の詳細の出力ペインに表示されます。 「タスクの実行履歴を表示する」を参照してください。