Använda aktivitetsvärden för att skicka information mellan aktiviteter

Aktivitetsvärden refererar till Databricks Utilities taskValues delverktyg, vilket gör att du kan skicka godtyckliga värden mellan uppgifter i ett Databricks-jobb. Se taskValues subutility (dbutils.jobs.taskValues).

Du anger ett nyckel/värde-par som använder dbutils.jobs.taskValues.set() i en aktivitet och kan sedan använda aktivitetsnamnet och nyckeln för att referera till värdet i efterföljande aktiviteter.

Ange uppgiftsvärden

Ange aktivitetsvärden i Python-notebook-filer med hjälp av dbutils.jobs.taskValues.set().

Nycklar för uppgiftsvärden måste vara strängar. Varje nyckel måste vara unik om du har flera aktivitetsvärden definierade i en notebook-fil.

Du kan manuellt eller programmatiskt tilldela aktivitetsvärden till nycklar. Endast värden som kan uttryckas som giltig JSON tillåts. Storleken på JSON-representationen av värdet får inte överstiga 48 KiB.

I följande exempel anges till exempel en statisk sträng för nyckeln fave_food:

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

Följande exempel använder en anteckningsboksuppgiftsparameter för att fråga alla uppdateringsposter för en viss bokning och returnera aktuell bokningsstatus och totalt antal poster:

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

Du kan skicka listor med värden med detta mönster och sedan använda dem för att samordna nedströms logik, såsom For each uppgifter. Se Använda en For each aktivitet för att köra en annan aktivitet i en loop.

Följande exempel extraherar de distinkta värdena för destinations-ID till en Python-lista och sätter detta som ett uppgiftsvärde:

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)

referensaktivitetsvärden

Databricks rekommenderar att du refererar till aktivitetsvärden som aktivitetsparametrar som konfigurerats med hjälp av referensmönstret för dynamiskt värde {{tasks.<task_name>.values.<value_name>}}.

Om du till exempel vill referera till aktivitetsvärdet med nyckeln dest_list från en aktivitet med namnet destination_lookupanvänder du syntaxen {{tasks.destination_lookup.values.dest_list}}.

Se Konfigurera aktivitetsparametrar och referenser för dynamiskt värde.

Använd dbutils.jobs.taskValues.get

Syntaxen dbutils.jobs.taskValues.get() kräver att du anger det överordnade aktivitetsnamnet. Den här syntaxen rekommenderas inte eftersom du kan använda aktivitetsvärden i flera underordnade aktiviteter, vilket innebär att många uppdateringar krävs om ett aktivitetsnamn ändras.

Med den här syntaxen kan du ange ett default värde och en debugValue. Standardvärdet används om nyckeln inte kan hittas. Med debugValue kan du ange ett statiskt värde som ska användas under manuell kodutveckling och testning i notebook-filer innan du schemalägger notebook-filen som en uppgift.

I följande exempel hämtas värdet för nyckeln booking_status som är inställt i ett uppgiftsnamn booking_lookup. Värdet confirmed returneras endast när notebook-filen körs interaktivt.

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

Läs värden från iterationerna i en For each aktivitet

En uppgift som körs i en För varje uppgift körs en gång per iteration, och varje iteration kan sätta ett uppgiftsvärde med dbutils.jobs.taskValues.set() som i vilken uppgift som helst. En nedströms uppgift kan läsa varje iterations värde från en enda returnerad lista, ordnad efter iterationsindex.

På båda läsytorna använder du nyckeln för den nästlade uppgiften som körs inuti For each uppgiften, inte själva For each uppgiften. Dess nyckel returnerar värdena från varje iteration som en enda lista ordnad efter iterationsindex. Till exempel, om tre iterationer av en nästlad uppgift med namnet process var och en anger result:

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

Den dynamiska värdereferensen {{tasks.process.values.result}} löses till samma array.

En iteration som inte satte nyckeln behåller sin position i resultatet: den visas som None (JSON null) från dbutils.jobs.taskValues.get, och som en null i en dynamisk värdereferens. Till exempel, om den andra av tre iterationer inte satte result, results är [1, None, 3].

Aggregeringsreferenser har två begränsningar: en sammansatt matris får inte överstiga cirka 48 KB (49 344 tecken) för ett enda parametervärde, och ett enda parametervärde kan innehålla högst tre aggregerade uppgiftsvärdereferenser.

Visa aktivitetsvärden

Det returnerade värdet för ett aktivitetsvärde för varje körning visas i fönstret Utdata i aktivitetskörningsinformationen. Se Visa aktivitetskörningshistorik.