Использование значений задач для передачи информации между задачами

Значения задачи относятся к утилитам Databricks, taskValues вспомогательный компонент которых позволяет передавать произвольные значения между задачами в задании Databricks. См. служебную программу taskValues (dbutils.jobs.taskValues).

Вы указываете пару «ключ — значение» с помощью dbutils.jobs.taskValues.set() в одной задаче, а затем можете использовать имя задачи и ключ для обращения к значению в последующих задачах.

Примечание.

Поскольку dbutils.jobs.taskValues.set() и dbutils.jobs.taskValues.get() в dbutils.jobs.taskValues являются функциями Python, они могут использоваться только в ноутбуках с Python, выбранным в качестве языка. Однако можно ссылаться на значения задач с помощью динамических ссылок на значения для всех задач, поддерживающих параметры. См. справочные значения задач.

Установить значения задачи

Задайте значения задач в записных книжках Python с помощью dbutils.jobs.taskValues.set().

Ключи значений задачи должны быть строками. Каждый ключ должен быть уникальным, если в записной книжке определено несколько значений задач.

Можно вручную или программно назначать значения задач ключам. Разрешены только значения, которые могут быть выражены как допустимые JSON. Размер JSON-представления значения не может превышать 48 КиБ.

Например, следующий пример задает статическую строку для ключа fave_food:

dbutils.jobs.taskValues.set(key = "fave_food", value = "beans")

В следующем примере используется параметр задачи notebook для запроса всех обновленных записей по конкретному бронированию и возврата текущего статуса бронирования и общего количества записей:

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 и задаёт их как значение задачи:

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 task, выполняется один раз в каждой итерации, и в каждой итерации можно задать значение задачи с помощью dbutils.jobs.taskValues.set(), как и в любой другой задаче. Наступающая задача может считывать значение каждой итерации из одного возвращаемого списка, упорядоченной по индексу итераций.

Important

Чтение агрегированных выходных данных итераций задачи For each доступно в бета-версии. Он включен по умолчанию и требует Databricks Runtime 15.4 LTS и выше.

На любой из поверхностей чтения используйте ключ вложенной задачи, выполняющейся внутри For each задачи, а не саму For each задачу. Его ключ возвращает значения каждой итерации как единый список, упорядоченный по индексу итераций. Например, если три итерации вложенной задачи с именем process каждая задают result:

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 КБ (49 344 символа) для одного значения параметра, а одно значение параметра может содержать не более трёх агрегированных ссылок на задачи.

Просмотр значений задач

Возвращаемое значение значения задачи для каждого запуска отображается в области выводасведений о выполнении задачи. См. Просмотр журнала выполнения задач.