Usar valores de tarefa para passar informações entre tarefas

Valores de tarefa referem-se ao subutilitário Databricks Utilities taskValues, que permite passar valores arbitrários entre tarefas numa tarefa Databricks. Consulte o subutilitário taskValues (dbutils.jobs.taskValues).

Você especifica um par chave-valor usando dbutils.jobs.taskValues.set() em uma tarefa e, em seguida, pode usar o nome da tarefa e a chave para fazer referência ao valor em tarefas subsequentes.

Definir valores de tarefa

Defina valores de tarefas em notebooks Python usando dbutils.jobs.taskValues.set().

As chaves de valor da tarefa devem ser cadeias de caracteres. Cada chave deve ser exclusiva se você tiver vários valores de tarefa definidos em um bloco de anotações.

Você pode atribuir manualmente ou programaticamente valores de tarefa a chaves. Somente valores que podem ser expressos como JSON válidos são permitidos. O tamanho da representação JSON do valor não pode exceder 48 KiB.

Por exemplo, o exemplo a seguir define uma cadeia de caracteres estática para a chave fave_food:

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

O exemplo seguinte utiliza um parâmetro de tarefa de notebook para consultar todos os registos de atualização de uma determinada reserva e devolver o estado atual da reserva e a contagem total de registos:

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])

Pode passar listas de valores usando este padrão e depois usá-las para coordenar a lógica subsequente, como as tarefas For each. Consulte Use uma For each tarefa para executar outra tarefa num ciclo.

O exemplo seguinte extrai os valores distintos do ID de destino para uma lista Python e define este valor como tarefa:

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)

Valores de tarefa de referência

O Databricks recomenda referenciar valores de tarefa como parâmetros de tarefa configurados usando o padrão de referência de valor dinâmico {{tasks.<task_name>.values.<value_name>}}.

Por exemplo, para fazer referência ao valor da tarefa com a chave dest_list de uma tarefa chamada destination_lookup, use a sintaxe {{tasks.destination_lookup.values.dest_list}}.

Consulte Configurar parâmetros de tarefa e Referências dinâmicas de valores.

Utilizar o comando dbutils.jobs.taskValues.get

A sintaxe dbutils.jobs.taskValues.get() requer a especificação do nome da tarefa upstream. Essa sintaxe não é recomendada, pois você pode usar valores de tarefa em várias tarefas downstream, o que significa que várias atualizações são necessárias se o nome de uma tarefa for alterado.

Utilizando esta sintaxe, pode opcionalmente especificar um valor default e um debugValue arquivo. O valor padrão é usado se a chave não puder ser encontrada. O debugValue permite definir um valor estático para usar durante o desenvolvimento manual de código e testes em blocos de anotações antes de agendar o bloco de anotações como uma tarefa.

O exemplo a seguir obtém o valor para a chave booking_status definida em um nome de tarefa booking_lookup. O valor confirmed é retornado somente ao executar o bloco de anotações interativamente.

booking_status = dbutils.jobs.taskValues.get(taskKey = "booking_lookup", key = "booking_status", debugValue = "confirmed")

Leia valores das iterações de uma For each tarefa

Uma tarefa executada dentro de Para cada é executada uma vez em cada iteração, e cada iteração pode definir um valor para a tarefa com dbutils.jobs.taskValues.set(), como em qualquer tarefa. Uma tarefa a jusante pode ler o valor de cada iteração a partir de uma única lista devolvida, ordenada por índice de iteração.

Em qualquer das superfícies de leitura, use a chave da tarefa aninhada que é executada dentro da tarefa For each, não da própria tarefa For each. A sua chave devolve os valores de cada iteração como uma única lista ordenada pelo índice de iteração. Por exemplo, se três iterações de uma tarefa aninhada com o nome process definirem, cada uma, result:

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

A referência de valor dinâmico {{tasks.process.values.result}} corresponde à mesma matriz.

Uma iteração que não definiu a chave mantém a sua posição no resultado: aparece como None (JSON null) a partir de dbutils.jobs.taskValues.get, e como a null numa referência de valor dinâmico. Por exemplo, se a segunda de três iterações não definiu result, results é [1, None, 3].

As referências agregadas têm dois limites: um array montado não pode exceder aproximadamente 48 KB (49.344 caracteres) para um único valor de parâmetro, e um único valor de parâmetro pode conter no máximo três referências agregadas de tarefa-valor.

Ver valores de tarefas

O valor devolvido do valor da tarefa em cada execução é apresentado no painel Saída dos detalhes da execução da tarefa. Consulte Ver histórico de execução de tarefas.