Använda en kontrolltabell för att köra ett For each jobb

När du kör samma bearbetning över många indata, såsom marknader, källtabeller, kunder eller datumpartitioner, innebär hårdkodning av listan i ditt jobb att redigera koden och distribuera om varje gång listan ändras. Istället lagrar du listan i en kontrolltabell som jobbet läser vid körning. För att lägga till eller ta bort arbete uppdaterar du en rad i tabellen, och nästa jobbkörning plockar upp ändringen utan ändringar i själva jobbet. Detta är ett metadatadrivet mönster: datan, inte koden, styr vad jobbet behandlar.

Denna handledning bygger ett jobb som använder detta mönster på den förinstallerade Wanderbricks-exempeldatasetet, så att du kan köra det från början till slut utan att skapa någon källdata. Scenariot är en semesteruthyrningsplattform som kör samma prisanalys för varje fastighetssegment (såsom Ski Resort eller Urban Year-Round). En kontrolltabell listar segmenten som ska analyseras, en SQL-uppgift läser den tabellen, och en For each uppgift kör analysen en gång per segment, parallellt.

Så här fungerar det

Jobbet kopplar ihop tre uppgifter i ordning:

Flödet är read_segments → process_segments → run_segment_analysis (en gång per varv). SQL-uppgiftens utdata, en JSON-array av radobjekt, flödar in i For each uppgiftens Inputs-fält via den dynamiska värdereferensen {{tasks.read_segments.output.rows}}. Uppgiften For each skickar sedan varje rads fält till den nästlade uppgiften som parametrar, tillgängliga som {{input.property_type}} och {{input.min_price}}.

Förutsättningar

• En Azure Databricks workspace med behörighet att skapa jobb och notebooks.
• Behörighet att skapa tabeller i Unity-katalogen, och behörighet att skapa ett schema i en katalog (och USE CATALOGCREATE SCHEMA privilegier) för att hålla kontrolltabellen.
• Ett SQL-lager för att köra SQL-uppgifterna. Om du inte har något, se Create a SQL warehouse.
• Katalogen samples , som finns tillgänglig i varje Unity Catalog-aktiverad arbetsyta. Handledningen läser från samples.wanderbricks.properties, så det finns ingen källdata att ställa upp.

Steg 1: Skapa kontrolltabellen

Kontrolltabellen är sanningskällan för listan över segment som ditt jobb hanterar. För att ändra vad jobbet gör uppdaterar du denna tabell, inte jobbet.

Kör följande SQL i en Azure Databricks-notebook eller SQL-editorn. Det första påståendet skapar ett schema för att hålla kontrolltabellen, och det andra skapar tabellen med en rad per fastighetssegment och det minsta listpriset att inkludera i segmentets analys:

USE CATALOG <catalog-name>;

CREATE SCHEMA IF NOT EXISTS config;

CREATE OR REPLACE TABLE config.property_segments AS
SELECT * FROM VALUES
  ('Urban Year-Round', 150),
  ('Summer Getaway', 200),
  ('Ski Resort', 250)
AS t(property_type, min_price);

Ersätt <catalog-name> med en katalog där du kan skapa scheman, som din arbetsplatskatalog. Använd samma katalog överallt som handledningen refererar config.property_segmentstill, inklusive uppslagsfrågan i steg 3.

Efter detta steg config.property_segments innehåller tre rader, en per segment. Varje rad bär de två värden som jobbet skickar till varje iteration: att analysera property_type och golvet min_price att filtrera på.

Steg 2: Skriv analyslogiken

Den nästlade uppgiften inuti uppgiften For each körs en gång per rad i kontrolltabellen, och tar emot den radens och property_type som parametrarmin_price. Du kan skriva denna logik som en anteckningsboksuppgift eller en SQL-uppgift. Välj baserat på din affärslogik:

• Använd en notebook-uppgift när per-iterationslogiken behöver procedurkod, flera språk eller bibliotek (till exempel ett data science- eller maskininlärningssteg).
• Använd en SQL-uppgift när logiken är en enda fråga eller transformation som du kan uttrycka deklarativt. En SQL-uppgift kräver ett SQL-lager.

Båda varianterna nedan ger samma resultat: för det segment som bearbetas, antalet listningar på eller över dess prisgolv samt deras genomsnittspris.

Notebook-uppgift

Skapa en ny notebook-fil på en sökväg, till exempel /Workspace/Users/<username>/run_segment_analysis. Denna anteckningsbok körs en gång per iteration av uppgiften For each och får ett nytt segment varje gång.

Lägg till följande kod i notebook-filen:

# Set default values so you can run the notebook on its own while developing.
# When the notebook runs inside a For each task, the job overrides these defaults.
dbutils.widgets.text("property_type", "Ski Resort", "Property type")
dbutils.widgets.text("min_price", "250", "Minimum price")

# Read the parameters passed by the For each task.
property_type = dbutils.widgets.get("property_type")
min_price = dbutils.widgets.get("min_price")

result = spark.sql(
    """
    SELECT :property_type AS property_type,
           COUNT(*) AS property_count,
           ROUND(AVG(base_price), 2) AS avg_price
    FROM samples.wanderbricks.properties
    WHERE property_type = :property_type
      AND base_price >= :min_price
    """,
    args={"property_type": property_type, "min_price": min_price},
)
display(result)

SQL-uppgift

En SQL-uppgift kör en sparad fråga, så skapa och spara analysfrågan i SQL-editorn nu. Du kopplar den till den nästlade uppgiften när du konfigurerar uppgiften For each i steg 4.

1. I din Azure Databricks workspace, klicka på Nytt>Fråga för att öppna SQL-editorn.

2. Ange följande fråga. SQL-uppgifter refererar parametrar med syntaxen :param_name , så att frågan läser sitt segment och prisgolv från och-parametrarna :property_type:min_price :

   SELECT :property_type AS property_type,
          COUNT(*) AS property_count,
          ROUND(AVG(base_price), 2) AS avg_price
   FROM samples.wanderbricks.properties
   WHERE property_type = :property_type
     AND base_price >= :min_price;
   

3. Klicka på titeln New Query <date> i flikrubriken på din SQL-fil och ge den namnet run_segment_analysis. Klicka sedan på Spara för att flytta den till en mapp där du vill lagra den.

Uppgiften For each skickar varje iterations värden till och :property_type namngivna :min_price parametrar vid körning. Till skillnad från notebook-widgets stöder SQL-namngivna parametrar inte standardvärden: om en parameter inte skickas misslyckas frågan med ett misstag i parameterupplösning.

Steg 3: Skapa uppslagsfrågan

Uppslagsuppgiften läser kontrolltabellen genom en sparad fråga. Som i steg 2, skapa och spara frågan i SQL-redigeraren nu, och bifoga den sedan till uppslagsuppgiften i steg 4.

1. I din Azure Databricks workspace, klicka på Nytt>Fråga för att öppna SQL-editorn.

2. Ange följande, med samma katalog som du valde i steg 1:

   SELECT property_type, min_price FROM <catalog-name>.config.property_segments;
   

   Namnet är fullt kvalificerat eftersom SQL-lagret som kör denna fråga kan gå till en annan katalog än den du skapade tabellen i.

3. Klicka på titeln New Query <date> i flikrubriken på din SQL-fil och ge den namnet read_segments. Klicka sedan på Spara för att flytta den till en mapp där du vill lagra den.

Steg 4: Skapa och konfigurera jobbet

Med båda frågorna sparade, skapa jobbet och lägg till dess två uppgifter: SQL-uppslagsuppgiften som läser kontrolltabellen och uppgiften For each som kör analysen för varje rad.

Skapa jobbet

I din Azure Databricks-arbetsyta, klicka på Nytt>Jobb. Ge jobbet ett beskrivande namn, som Segment Analysis.

Konfigurera SQL-uppslagsuppgiften

Denna uppgift läser kontrolltabellen och gör dess rader tillgängliga för uppgiften For each genom att köra den read_segments fråga du sparade i steg 3.

1. Klicka på SQL-frågerutan för att konfigurera den första uppgiften. Om SQL-frågerutan inte är tillgänglig, klicka på Lägg till en annan uppgiftstyp och sök efter SQL-fråga.
2. Ange Aktivitetsnamn till read_segments.
3. Om det behövs, välj SQL-fråga i rullgardinsmenyn för Typ .
4. I SQL-frågefältet väljer du den read_segments fråga du sparade i steg 3.
5. Ställ in SQL Warehouse till ett datavaruhus på din arbetsyta.
6. Klicka på Skapa uppgift.

När denna uppgift körs fångar Azure Databricks resultatet som en JSON-array i tasks.read_segments.output.rows. SQL-uppgiftsutdata returneras alltid som en JSON-array, så du behöver ingen extra konfiguration. Den allmänna formen av referensen är tasks.<task-name>.output.rows, där <task-name> matchar uppgiftsnamnet du ställt in. Utdata ser ut så här:

[
  { "property_type": "Urban Year-Round", "min_price": 150 },
  { "property_type": "Summer Getaway", "min_price": 200 },
  { "property_type": "Ski Resort", "min_price": 250 }
]

Konfigurera uppgiften For each

Uppgiften For each läser SQL-utdata och startar en kapslad aktivitetskörning per rad.

1. Klicka Lägg till uppgift och välj För varje.

2. Ange Aktivitetsnamn till process_segments.

3. Kontrollera att Depends på är satt till read_segments.

4. I fältet Inputs , ange radarrayen som fångas av SQL-uppgiften:

   {{tasks.read_segments.output.rows}}
   

5. Ställ in Concurrency på 2 att köra två iterationer parallellt. Öka det här värdet när din kapslade uppgift stöder högre parallellitet.

6. För att slutföra denna uppgift, klicka på Lägg till en uppgift för att loopa över och konfigurera den nästlade uppgift som körs på varje iteration.

Uppgiften For each och dess nästlade uppgift skapas tillsammans som en enda uppgift. Konfigurera den nästlade uppgiften baserat på den typ du valde i steg 2:

Notebook-uppgift

1. Ange Aktivitetsnamn till run_segment_analysis.

2. Ställ in Typ på Notebook.

3. Sätt Path till anteckningsboken du skapade i steg 2.

4. Klicka på Parametrar, sedan på Lägg till för att lägga till varje parameter:

   • Nyckel: property_type, Värde: {{input.property_type}}
   • Nyckel: min_price, Värde: {{input.min_price}}

   Varje {{input.<key>}} referens löses till matchningsfältet från raden i den aktuella iterationen.

5. Klicka på Skapa uppgift för att skapa For each uppgiften och dess inbäddade uppgift tillsammans.

SQL-uppgift

Denna uppgift kör den run_segment_analysis fråga du sparade i steg 2.

1. Ange Aktivitetsnamn till run_segment_analysis.

2. Sätt Type till SQL, och sätt sedan SQL-uppgiften till Query.

3. I SQL-frågefältet , välj den run_segment_analysis fråga du sparade i Steg 2.

4. Ställ in SQL Warehouse till ett datavaruhus på din arbetsyta.

5. Klicka på Parametrar, sedan på Lägg till för att lägga till varje parameter:

   • Nyckel: property_type, Värde: {{input.property_type}}
   • Nyckel: min_price, Värde: {{input.min_price}}

   Varje {{input.<key>}} referens löses till matchningsfältet från raden i den aktuella iterationen.

6. Klicka på Skapa uppgift för att skapa For each uppgiften och dess inbäddade uppgift tillsammans.

Din jobb Directed Acyclic Graph (DAG) visar read_segments nu att den flödar in i process_segments, med den nästlade uppgiften inuti For each noden.

Steg 5: Kör jobbet och verifiera

1. Klicka på Kör nu för att utlösa jobbet.
2. Välj fliken Runs för att se runen. Den första körningen av ett jobb tar några minuter att starta beräkningen; När den är klar visas den i listan.
3. Klicka på process_segments noden för att expandera uppgiften For each .
4. Körsidan visar en tabell över iterationer, en rad per segment, var och en med sin status, starttid och varaktighet.
5. Klicka på en iterationsrad för att öppna dess utdata och bekräfta att den analyserat det förväntade segmentet.

Du kan se resultaten av varje iteration oberoende av varandra. Om en specifik iteration misslyckas kan du köra om endast den iterationen från jobbets körsida utan att köra hela jobbet igen.

Utöka mönstret

För att lägga till ett segment i analysen, infoga en rad i kontrolltabellen:

INSERT INTO <catalog-name>.config.property_segments VALUES ('Historical Place', 100);

Nästa jobbkörning inkluderar det nya segmentet, utan ändringar i jobbkonfiguration eller redigeringar i anteckningsboken.

Samma mönster gäller i alla fall där du vill att data ska driva iteration:

• Bearbetning per kund: En rad per kund-ID. Den nästlade uppgiften tillämpar kundspecifika transformationer eller levererar till kundspecifika destinationer.
• Tabellinmatning: En rad per källtabellnamn. Den inbäddade uppgiften läser och tar in varje tabell.
• Återfyllnadsbearbetning: En rad per datumpartition. Den nästlade uppgiften bearbetar historisk data för den partitionen.
• Körning styrd av funktionsflaggor: En rad per aktiverad funktion eller experiment. Den nästlade uppgiften aktiverar motsvarande logik.

För att sluta bearbeta en rad utan att ta bort den, lägg till din egen kolumn i kontrolltabellen (som en active flagga) och filtrera på den i SQL-uppslagsuppgiften. Detta är en vanlig kolumn som du definierar och fyller i; Uppgiften For each har inget inbyggt koncept. Lägg först till kolumnen, och sätt sedan de befintliga raderna till TRUE:

ALTER TABLE <catalog-name>.config.property_segments ADD COLUMN active BOOLEAN;
UPDATE <catalog-name>.config.property_segments SET active = TRUE;

Filtrera sedan på den i read_segments frågan så att endast aktiva rader styr iterationen:

SELECT property_type, min_price FROM <catalog-name>.config.property_segments WHERE active = TRUE;

Ytterligare resurser

• Använd en For each aktivitet för att köra en annan aktivitet i en loop: Fullständig referens för att For each konfigurera uppgifter, inklusive parametertyper och samtidighetsalternativ
• Använd en uppslagstabell för stora parametermatriser i en For each uppgift: Hantera stora parametermatriser som överskrider aktivitetsvärdegränsen på 48 KB
• Åtkomst till parametervärden från en aktivitet: Alla metoder för att komma åt parametervärden i notebook-filer, Python skript och SQL-uppgifter
• Wanderbricks-dataset: Exempeldataset som används i denna handledning