使用任务值在任务之间传递信息

任务值是指 Databricks 实用程序 taskValues 的子实用工具，你可以用它在 Databricks 作业中的任务之间传递任意值。 请参阅 taskValues 子实用工具 (dbutils.jobs.taskValues)。

可以在一个任务中使用 dbutils.jobs.taskValues.set() 指定键值对，然后使用任务名称和键引用后续任务中的值。

设置任务值

使用 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")

从 For each 任务的迭代中读取值

在 对每个任务 中运行的任务会在每次迭代中运行一次，并且每次迭代都可以像在任何任务中一样使用 dbutils.jobs.taskValues.set() 设置任务值。 下游任务可以从单个返回列表中读取每次迭代的值，列表按迭代索引排序。

在任一读取端上，请使用在 For each 任务内运行的嵌套任务的键，而不是 For each 任务本身的键。 其键将每次迭代的值以单一列表的形式返回，按迭代索引排序。 例如，如果某个名为 result 的嵌套任务进行了三次迭代，且每次迭代都设置了 process：

results = dbutils.jobs.taskValues.get(taskKey = "process", key = "result")
# results == [1, 2, 3]

动态值引用 {{tasks.process.values.result}} 解析为相同的数组。

未设置键的迭代会保持其在结果中的位置：它作为 None （JSON null）从 dbutils.jobs.taskValues.get中出现，在动态值引用中以a null 形式出现。 例如，如果三次迭代中的第二次没有设置 result， results 则 是 [1, None, 3]。

聚合引用有两个限制：汇编后的数组单个参数值最多不能超过约48 KB（49,344字符），单个参数值最多包含三个聚合任务值引用。

查看任务值

每个运行的任务值的返回值显示在任务运行详细信息的“输出”窗格中。 请参阅查看任务运行历史记录。