Taakwaarden gebruiken om informatie door te geven tussen taken

taakwaarden verwijzen naar de Databricks Utilities taskValues subutiliteit, waarmee u willekeurige waarden tussen taken in een Databricks-taak kunt doorgeven. Zie het hulpprogramma taskValues (dbutils.jobs.taskValues).

U geeft een sleutel-waardepaar op met behulp van dbutils.jobs.taskValues.set() in een taak en kunt vervolgens de taaknaam en sleutel gebruiken om naar de waarde te verwijzen in daaropvolgende taken.

Taakwaarden instellen

Stel taakwaarden in Python-notebooks in met behulp van dbutils.jobs.taskValues.set().

Sleutels van taakwaarden moeten tekenreeksen zijn. Elke sleutel moet uniek zijn als u meerdere taakwaarden hebt gedefinieerd in een notebook.

U kunt taakwaarden handmatig of programmatisch toewijzen aan sleutels. Alleen waarden die als geldige JSON kunnen worden uitgedrukt, zijn toegestaan. De grootte van de JSON-weergave van de waarde mag niet groter zijn dan 48 KiB.

In het volgende voorbeeld wordt bijvoorbeeld een statische tekenreeks ingesteld voor de sleutel fave_food:

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

Het volgende voorbeeld gebruikt een notebook-taakparameter om alle updaterecords voor een bepaalde boeking op te vragen en de huidige boekingsstatus en het totale aantal records terug te geven:

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

Je kunt lijsten met waarden doorgeven met dit patroon en ze vervolgens gebruiken om downstreamlogica, zoals For each taken, te coördineren. Zie Een For each taak gebruiken om een andere taak in een lus uit te voeren.

Het volgende voorbeeld haalt de verschillende waarden voor bestemmings-ID uit naar een Python-lijst en stelt dit in als een taakwaarde:

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)

Referentietaakwaarden

Databricks raadt aan om te verwijzen naar taakwaarden als taakparameters die zijn geconfigureerd met behulp van het referentiepatroon voor dynamische waarden {{tasks.<task_name>.values.<value_name>}}.

Als u bijvoorbeeld wilt verwijzen naar de taakwaarde met de sleutel dest_list van een benoemde destination_lookuptaak, gebruikt u de syntaxis {{tasks.destination_lookup.values.dest_list}}.

Zie Taakparameters configureren en Dynamische waardeverwijzingen.

dbutils.jobs.taskValues.get gebruiken

Voor de syntaxis dbutils.jobs.taskValues.get() moet de naam van de upstream-taak worden opgegeven. Deze syntaxis wordt niet aanbevolen, omdat u taakwaarden in meerdere downstreamtaken kunt gebruiken, wat betekent dat er talloze updates nodig zijn als een taaknaam wordt gewijzigd.

Met deze syntaxis kunt u desgewenst een waarde voor default en een waarde voor debugValue opgeven. De standaardwaarde wordt gebruikt als de sleutel niet kan worden gevonden. Met de debugValue kunt u een statische waarde instellen die moet worden gebruikt tijdens het handmatig ontwikkelen en testen van code in notebooks voordat u het notebook als taak plant.

In het volgende voorbeeld wordt de waarde van de sleutel booking_status opgehaald, die is opgenomen in de taaknaam booking_lookup. De waarde confirmed wordt alleen geretourneerd wanneer u het notebook interactief uitvoert.

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

Lees waarden uit de iteraties van een For each taak

Een taak die binnen een Voor elke taak draait, draait één keer per iteratie, en elke iteratie kan een taakwaarde instellen met dbutils.jobs.taskValues.set() zoals in elke taak. Een downstream-taak kan de waarde van elke iteratie lezen uit een enkele teruggestuurde lijst, geordend op iteratie-index.

Gebruik in beide leesinterfaces de sleutel van de geneste taak die in de taak For each wordt uitgevoerd, niet die van de taak For each zelf. De sleutel ervan geeft de waarden van elke iteratie terug als één enkele lijst, geordend op iteratie-index. Bijvoorbeeld, als drie iteraties van een geneste taak met de naam result elk process instellen:

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

De dynamische waardereferentie {{tasks.process.values.result}} verwijst naar dezelfde array.

Een iteratie die de sleutel niet heeft ingesteld, behoudt zijn positie in het resultaat: deze verschijnt als None (JSON null) uit dbutils.jobs.taskValues.get, en als een null in een dynamische waardereferentie. Bijvoorbeeld, als de tweede van de drie iteraties result niet heeft ingesteld, is results[1, None, 3].

Geaggregeerde referenties hebben twee limieten: een geassembleerde array mag niet groter zijn dan ongeveer 48 KB (49.344 tekens) voor één parameterwaarde, en een enkele parameterwaarde mag maximaal drie geaggregeerde taakwaardereferenties bevatten.

Taakwaarden weergeven

De geretourneerde waarde van een taakwaarde voor elke uitvoering wordt weergegeven in het deelvenster Uitvoer van de details van de taakuitvoering. Zie De uitvoeringsgeschiedenis van de taak weergeven.