任務值是指 Databricks 公用程式 taskValues 子公用程式,這可讓您在 Databricks 工作中的任務之間傳遞任意值。 請參閱 taskValues 子公用程式 (dbutils.jobs.taskValues)。
您可以在一個任務中使用 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 在迴圈中執行另一個任務。
以下範例將目的 ID 的不同值提取到 Python 清單,並將其設為任務值:
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
在 For each 工作 內執行的工作,於每次反覆運算中會執行一次,而每次反覆運算都可像在任何工作中一樣使用 dbutils.jobs.taskValues.set() 設定工作值。 下游任務可以從一個以迭代索引排序的單一回傳清單讀取每次迭代的值。
Important
讀取 For each 任務的彙整迭代輸出仍處於 Beta 階段。 預設啟用,且需 Databricks Runtime 15.4 LTS 或以上版本。
在任一讀取介面上,請使用在 For each 任務內執行的巢狀任務之鍵,而不是 For each 任務本身。 其鍵以單一清單形式回傳每次迭代的值,並依迭代索引排序。 例如,若名為 result 的巢狀任務三次迭代時,每次都設定 process:
results = dbutils.jobs.taskValues.get(taskKey = "process", key = "result")
# results == [1, 2, 3]
動態值參考 {{tasks.process.values.result}} 會解析到相同的陣列。
未設定索引鍵的迭代會在結果中保留其位置:它會以 null(JSON 中為 dbutils.jobs.taskValues.get)的形式顯示,來自 null,並在動態值參照中顯示為 None。 例如,若三次迭代中的第二次未設定 result, results 則為 [1, None, 3]。
聚合後的參考有兩個限制:一個組合後的陣列對單一參數值的容量不得超過約 48 KB(49,344 個字元),而單一參數值最多只能包含三個聚合後的任務值參考。
檢視任務值
每次執行的任務回傳值會顯示在任務執行細節的輸出面板中。 請參閱檢視任務執行歷程記錄。