Verwenden von Aufgabenwerten zur Weitergabe von Informationen zwischen Aufgaben

Aufgabenwerte beziehen sich auf das Unterhilfsprogramm taskValues von Databricks Utilities, mit der Sie beliebige Werte zwischen Aufgaben in einem Databricks-Job übergeben können. Weitere Informationen finden Sie unter taskValues-Unterhilfsprogramm (dbutils.jobs.taskValues).

Sie geben ein Schlüssel-Wert-Paar in einer Aufgabe mit dbutils.jobs.taskValues.set() an und können dann den Aufgabennamen und den Schlüssel verwenden, um in nachfolgenden Aufgaben auf den Wert zu verweisen.

Aufgabenwerte festlegen

Legen Sie Werte für Aufgaben in Python-Notizbüchern mithilfe von dbutils.jobs.taskValues.set() fest.

Aufgabenwertschlüssel müssen Zeichenketten sein. Jeder Schlüssel muss eindeutig sein, wenn mehrere Aufgabenwerte in einem Notebook definiert sind.

Sie können Aufgabenwerte manuell oder programmgesteuert Schlüsseln zuweisen. Es sind nur Werte zulässig, die als gültiges JSON ausgedrückt werden können. Die Größe der JSON-Darstellung des Werts darf 48 KiB nicht überschreiten.

Im folgenden Beispiel wird beispielsweise eine statische Zeichenfolge für den Schlüssel fave_foodfestgelegt:

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

Das folgende Beispiel verwendet einen Notebook-Task-Parameter, um alle Aktualisierungsdatensätze für eine bestimmte Buchung abzufragen und den aktuellen Buchungsstatus sowie die Gesamtanzahl der Datensätze zurückzugeben:

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

Man kann Wertelisten mit diesem Muster übergeben und sie dann zur Koordination der nachgelagerten Logik wie For each Aufgaben verwenden. Siehe Verwenden Sie eine Aufgabe, um eine andere For each Aufgabe in einer Schleife auszuführen.

Das folgende Beispiel extrahiert die unterschiedlichen Werte für die Ziel-ID in eine Python-Liste und setzt diese als Aufgabenwert:

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)

Referenzwerte für Aufgaben

Databricks empfiehlt das Verweisen auf Aufgabenwerte als Aufgabenparameter, die mithilfe des dynamischen Wertvreferenzmusters {{tasks.<task_name>.values.<value_name>}} konfiguriert sind.

Um zum Beispiel den Aufgabenwert mit dem Schlüssel dest_list einer Aufgabe mit dem Namen destination_lookup zu referenzieren, verwenden Sie die Syntax {{tasks.destination_lookup.values.dest_list}}.

Siehe Konfigurieren von Vorgangsparametern und Dynamischen Wertverweise.

Verwenden Sie dbutils.jobs.taskValues.get

Für die Syntax dbutils.jobs.taskValues.get() ist die Angabe des Upstream-Aufgabennamens erforderlich. Diese Syntax wird nicht empfohlen, da Sie Aufgabenwerte in mehreren nachgelagerten Aufgaben verwenden können, was bedeutet, dass zahlreiche Aktualisierungen erforderlich sind, wenn sich ein Aufgabenname ändert.

Mithilfe dieser Syntax können Sie optional einen default-Wert und ein debugValue festlegen. Der Standardwert wird verwendet, wenn der Schlüssel nicht gefunden werden kann. debugValue ermöglicht es Ihnen, einen statischen Wert festzulegen, der während der manuellen Codeentwicklung und beim Testen in Notebooks verwendet werden soll, bevor Sie das Notebook als Aufgabe planen.

Im folgenden Beispiel wird der Wert für den Schlüssel booking_status abgerufen, der in einem Aufgabennamen booking_lookup festgelegt ist. Der Wert confirmed wird nur zurückgegeben, wenn das Notebook interaktiv ausgeführt wird.

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

Lesen Sie Werte aus den Iterationen einer For each Aufgabe

Eine Aufgabe, die innerhalb von Für jede Aufgabe ausgeführt wird, läuft einmal pro Iteration, und jede Iteration kann einen Aufgabenwert mit dbutils.jobs.taskValues.set() wie in jeder beliebigen Aufgabe festlegen. Eine nachgelagerte Aufgabe kann den Wert jeder Iteration aus einer einzelnen zurückgegebenen Liste ablesen, geordnet nach Iterationsindex.

Auf beiden Leseseiten verwenden Sie den Schlüssel der verschachtelten Aufgabe, die innerhalb der Aufgabe For each ausgeführt wird, nicht den der Aufgabe For each selbst. Sein Schlüssel gibt die Werte jeder Iteration als eine einzige Liste zurück, geordnet nach Iterationsindex. Wenn beispielsweise drei Iterationen einer verschachtelten Aufgabe mit dem Namen result jeweils process festlegen:

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

Die dynamische Wertreferenz {{tasks.process.values.result}} wird zu demselben Array aufgelöst.

Eine Iteration, die den Schlüssel nicht gesetzt hat, behält ihre Position im Ergebnis: Sie erscheint als None (JSON null) aus dbutils.jobs.taskValues.get, und als a null in einer dynamischen Wertreferenz. Wenn beispielsweise in der zweiten von drei Iterationen result nicht festgelegt wurde, ist results[1, None, 3].

Aggregierte Referenzen haben zwei Grenzen: Ein zusammengesetztes Array darf für einen einzelnen Parameterwert nicht mehr als etwa 48 KB (49.344 Zeichen) überschreiten, und ein einzelner Parameterwert darf höchstens drei aggregierte Aufgabenwertreferenzen enthalten.

Anzeigen von Aufgabenwerten

Der zurückgegebene Wert eines Vorgangswerts für jede Ausführung wird im Ausgabebereich der Aufgabenausführungsdetails angezeigt. Weitere Informationen finden Sie unter Anzeigen des Ausführungsverlaufs von Aufgaben.