Verwende eine Steuertabelle, um einen For each Auftrag zu steuern

Wenn Sie dieselbe Verarbeitung über viele Eingaben ausführen, wie Märkte, Quelltabellen, Kunden oder Datumspartitionen, bedeutet das Hardcodieren dieser Liste in Ihrem Job, Code zu bearbeiten und jedes Mal neu zu deployen, wenn sich die Liste ändert. Speichere stattdessen die Liste in einer Steuertabelle , die der Job zur Laufzeit liest. Um Arbeit hinzuzufügen oder zu entfernen, aktualisiert man eine Zeile in der Tabelle, und der nächste Job-Run übernimmt die Änderung ohne Änderungen am Job selbst. Dies ist ein metadatengetriebenes Muster: Die Daten, nicht der Code, steuern, was der Job verarbeitet.

Dieses Tutorial erstellt einen Job, der dieses Muster auf dem vorinstallierten Wanderbricks-Beispieldatensatz verwendet, sodass man es von Ende zu Ende ausführen kann, ohne Quelldaten zu erstellen. Das Szenario ist eine Plattform für Ferienvermietungen, die für jedes Immobiliensegment die gleiche Preisanalyse durchführt (wie Ski Resort oder Urban Year-Round). Eine Kontrolltabelle listet die zu analysierenden Segmente auf, eine SQL-Aufgabe liest diese Tabelle, und eine For each Aufgabe führt die Analyse pro Segment einmal parallel durch.

Wie der Control-Table-Job funktioniert

Der Job verbindet drei Tasks der Reihe nach:

Der Ablauf ist read_segments → process_segments → run_segment_analysis (einmal pro Zeile). Die Ausgabe der SQL-Aufgabe, ein JSON-Array von Zeilenobjekten, fließt über die dynamische Wertreferenz For each in das Eingaben-Feld der Aufgabe {{tasks.read_segments.output.rows}}. Die Aufgabe For each übergibt dann die Felder jeder Zeile als Parameter an die verschachtelte Aufgabe, verfügbar als {{input.property_type}} und {{input.min_price}}.

Voraussetzungen

• Ein Azure Databricks-Arbeitsbereich mit Berechtigung zum Erstellen von Jobs und Notizbüchern.
• Berechtigung, Tabellen im Unity-Katalog zu erstellen, und Berechtigung, ein Schema in einem Katalog zu erstellen (die USE CATALOG und CREATE SCHEMA Privilegien), um die Kontrolltabelle zu halten.
• Ein SQL-Warehouse, um die SQL-Aufgaben auszuführen. Falls du keines hast, siehe Ein SQL-Warehouse erstellen.
• Der samples-Katalog, der in jedem von Unity Catalog aktivierten Arbeitsbereich verfügbar ist. Das Tutorial liest aus samples.wanderbricks.properties, also gibt es keine Quelldaten zum Einrichten.

Schritt 1: Erstellen Sie die Kontrolltabelle

Die Kontrolltabelle ist die maßgebliche Quelle für die Liste der Segmente, die Ihr Job verarbeitet. Um zu ändern, was der Job macht, aktualisiert man diese Tabelle, nicht den Job.

Führe folgendes SQL in einem Azure Databricks-Notebook oder im SQL-Editor aus. Die erste Anweisung erstellt ein Schema zur Aufnahme der Steuertabelle, und die zweite erstellt die Tabelle mit einer Zeile pro Eigenschaftssegment und dem Mindestlistenpreis, der in die Analyse dieses Segments aufgenommen werden soll:

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

Ersetzen Sie <catalog-name> durch einen Katalog, in dem Sie Schemata erstellen können, wie zum Beispiel Ihren Arbeitsbereichskatalog. Verwenden Sie überall denselben Katalog, auf den das Tutorial verweist config.property_segments, einschließlich der Suchanfrage in Schritt 3.

Nach diesem Schritt enthält config.property_segments drei Zeilen, eine pro Segment. Jede Zeile enthält die zwei Werte, die der Auftrag an jede Iteration weitergibt: die property_type zum Analysieren und die min_price-Etage, nach der gefiltert wird.

Schritt 2: Schreiben Sie die Analyselogik

Die verschachtelte Aufgabe innerhalb der For each Aufgabe wird pro Zeile der Kontrolltabelle einmal ausgeführt und erhält die Parameter dieser Zeile property_type und min_price als Parameter. Sie können diese Logik als Notebook-Aufgabe oder SQL-Aufgabe erstellen. Wählen Sie basierend auf Ihrer Geschäftslogik:

• Verwenden Sie eine Notebook-Aufgabe , wenn die Per-Iteration-Logik prozeduralen Code, mehrere Sprachen oder Bibliotheken benötigt (zum Beispiel einen Data-Science- oder Machine-Learning-Schritt).
• Verwenden Sie eine SQL-Aufgabe, wenn die Logik eine einzelne Abfrage oder Transformation ist, die Sie deklarativ ausdrücken können. Eine SQL-Aufgabe benötigt ein SQL-Warehouse.

Beide unten aufgeführten Varianten liefern das gleiche Ergebnis: Für das zu bearbeitende Segment werden die Anzahl der Angebote auf oder über der Preisuntergrenze und deren Durchschnittspreis ausgegeben.

Notebook-Aufgabe

Erstellen Sie ein neues Notizbuch unter einem Pfad wie /Workspace/Users/<username>/run_segment_analysis. Dieses Notebook läuft pro Iteration der For each-Aufgabe einmal und erhält jedes Mal ein anderes Segment.

Fügen Sie dem Notizbuch den folgenden Code hinzu:

# 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-Aufgabe

Eine SQL-Aufgabe führt eine gespeicherte Abfrage aus, also erstellen und speichern Sie die Analyseanfrage jetzt im SQL-Editor. Du hängst sie an die verschachtelte Aufgabe an, wenn du die For each Aufgabe in Schritt 4 konfigurierst.

1. Klicken Sie in Ihrem Azure Databricks-Arbeitsbereich auf Neu>Abfrage, um den SQL-Editor zu öffnen.

2. Geben Sie die folgende Abfrage ein. SQL-Aufgaben beziehen sich auf Parameter mit der Syntax :param_name, sodass die Abfrage ihr Segment und Preisboden aus den Parametern :property_type und :min_price abliest:

   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. Klicken Sie auf den Titel New Query <date> in der Registerkartenüberschrift Ihrer SQL-Datei und geben Sie ihr den Namen run_segment_analysis. Dann klicke auf Speichern , um es in einen Ordner zu verschieben, in dem du es speichern möchtest.

Die Aufgabe For each übergibt die Werte jeder Iteration zur Laufzeit an die :property_type und :min_price benannten Parameter. Im Gegensatz zu Notebook-Widgets unterstützen SQL-benannte Parameter keine Standardwerte: Wenn ein Parameter nicht übergeben wird, schlägt die Abfrage mit einem Parameter-Resolutionsfehler fehl.

Schritt 3: Erstellen Sie die Lookup-Abfrage

Der Lookup-Task liest die Steuertabelle über eine gespeicherte Abfrage. Wie in Schritt 2 erstelle und speichere die Abfrage jetzt im SQL-Editor und hänge sie dann in Schritt 4 an die Lookup-Aufgabe an.

1. Klicken Sie in Ihrem Azure Databricks-Arbeitsbereich auf Neu>Abfrage, um den SQL-Editor zu öffnen.

2. Geben Sie Folgendes ein, mit demselben Katalog, den Sie in Schritt 1 ausgewählt haben:

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

   Der Name ist vollständig qualifiziert, weil das SQL-Warehouse, das diese Abfrage ausführt, standardmäßig auf einen anderen Katalog zurückgreifen könnte als der, in dem Sie die Tabelle erstellt haben.

3. Klicken Sie auf den Titel New Query <date> in der Registerkartenüberschrift Ihrer SQL-Datei und geben Sie ihr den Namen read_segments. Dann klicke auf Speichern , um es in einen Ordner zu verschieben, in dem du es speichern möchtest.

Schritt 4: Erstellen und konfigurieren Sie den Job

Nachdem beide Abfragen gespeichert wurden, erstellen Sie den Auftrag und fügen seine zwei Aufgaben hinzu: die SQL-Abfrageaufgabe, die die Steuertabelle liest, und die For each-Aufgabe, die die Analyse für jede Zeile ausführt.

Job erstellen

In Ihrem Azure Databricks-Arbeitsbereich klicken Sie in der Seitenleiste auf Neu>Auftrag. Gib dem Job einen beschreibenden Namen, wie zum Beispiel Segment Analysis.

Konfigurieren Sie die SQL-Abfrageaufgabe

Diese Aufgabe liest die Kontrolltabelle und stellt deren Zeilen der For each-Aufgabe zur Verfügung, indem sie die in Schritt 3 gespeicherte read_segments-Abfrage ausführt.

1. Klicken Sie auf die SQL-Abfragekachel , um die erste Aufgabe zu konfigurieren. Wenn die SQL-Abfragekachel nicht verfügbar ist, klicken Sie auf Einen weiteren Aufgabentyp hinzufügen und suchen Sie nach SQL-Abfrage.
2. Legen Sie den Aufgabennamen auf read_segments.
3. Falls nötig, wählen Sie SQL-Abfrage aus dem Dropdown-Menü Typ aus.
4. Wählen Sie im SQL-Abfragefeld die read_segments in Schritt 3 gespeicherte Abfrage aus.
5. Stellen Sie das SQL-Warehouse auf ein Warehouse in Ihrem Arbeitsbereich ein.
6. Klicken Sie auf Aufgabe erstellen.

Wenn diese Aufgabe ausgeführt wird, erfasst Azure Databricks das Ergebnis als JSON-Array in tasks.read_segments.output.rows. Die SQL-Aufgabenausgabe wird immer als JSON-Array zurückgegeben, daher benötigen Sie keine zusätzliche Konfiguration. Die allgemeine Form der Referenz ist tasks.<task-name>.output.rows, wobei <task-name> mit dem von dir festgelegten Aufgabennamen übereinstimmt. Die Ausgabe sieht wie folgt aus:

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

Aufgabe For each konfigurieren

Die For each Aufgabe liest die SQL-Ausgabe und startet eine geschachtelte Aufgabe pro Zeile.

1. Klicken Sie auf Aufgabe hinzufügen, und wählen Sie Für jede aus.

2. Legen Sie den Aufgabennamen auf process_segments.

3. Überprüfen Sie, dass Depends on auf read_segments gesetzt ist.

4. Im Feld Inputs geben Sie das von der SQL-Aufgabe erfasste Zeilenarray ein:

   {{tasks.read_segments.output.rows}}
   

5. Stellen Sie Concurrency so ein 2 , dass zwei Iterationen parallel ausgeführt werden. Erhöhen Sie diesen Wert, wenn Ihre geschachtelte Aufgabe höhere Parallelität unterstützt.

6. Um diese Aufgabe abzuschließen, klicken Sie auf Aufgabe zum Durchlaufen hinzufügen und konfigurieren Sie die verschachtelte Aufgabe, die bei jeder Iteration ausgeführt wird.

Die For each Aufgabe und ihre verschachtelte Aufgabe werden zusammen als eine einzige Aufgabe erstellt. Konfigurieren Sie die verschachtelte Aufgabe basierend auf dem Typ, den Sie in Schritt 2 gewählt haben:

Notebook-Aufgabe

1. Legen Sie den Aufgabennamen auf run_segment_analysis.

2. Legen Sie "Typ " auf " Notizbuch" fest.

3. Setzen Sie Pfad auf das Notebook, das Sie in Schritt 2 erstellt haben.

4. Klicken Sie auf Parameter und dann auf Hinzufügen , um jeden Parameter hinzuzufügen:

   • Schlüssel: , property_type: {{input.property_type}}
   • Schlüssel: , min_price: {{input.min_price}}

   Jede {{input.<key>}}-Referenz wird aus der Zeile der aktuellen Iteration in das passende Feld aufgelöst.

5. Klicken Sie auf Aufgabe erstellen, um die For each-Aufgabe und ihre verschachtelte Aufgabe gemeinsam zu erstellen.

SQL-Aufgabe

Diese Aufgabe führt die run_segment_analysis in Schritt 2 gespeicherte Abfrage aus.

1. Legen Sie den Aufgabennamen auf run_segment_analysis.

2. Setzen Sie Type auf SQL, dann setzen Sie SQL-Aufgabe auf Abfrage.

3. Wählen Sie im SQL-Abfragefeld die run_segment_analysis in Schritt 2 gespeicherte Abfrage aus.

4. Stellen Sie das SQL-Warehouse auf ein Warehouse in Ihrem Arbeitsbereich ein.

5. Klicken Sie auf Parameter und dann auf Hinzufügen , um jeden Parameter hinzuzufügen:

   • Schlüssel: , property_type: {{input.property_type}}
   • Schlüssel: , min_price: {{input.min_price}}

   Jede {{input.<key>}}-Referenz wird aus der Zeile der aktuellen Iteration in das passende Feld aufgelöst.

6. Klicken Sie auf Aufgabe erstellen, um die For each-Aufgabe und ihre verschachtelte Aufgabe gemeinsam zu erstellen.

Dein Job Directed Acyclic Graph (DAG) zeigt jetzt, wie read_segments in process_segments fließt, mit der verschachtelten Aufgabe innerhalb des Knotens For each.

Schritt 5: Führen Sie den Auftrag durch und verifizieren Sie

1. Klicken Sie auf "Jetzt ausführen" , um den Auftrag auszulösen.
2. Wählen Sie die Registerkarte Ausführungen aus, um die Ausführung anzuzeigen. Die erste Ausführung eines Auftrags benötigt einige Minuten, um Rechenressourcen zu starten; wenn er abgeschlossen ist, erscheint er in der Liste.
3. Klicken Sie auf den process_segments Knoten, um die For each Aufgabe zu erweitern.
4. Die Laufseite zeigt eine Tabelle der Iterationen, eine Zeile pro Segment, jede mit Status, Startzeit und Dauer.
5. Klicken Sie auf eine beliebige Iterationszeile, um deren Ausgabe zu öffnen und zu bestätigen, dass das erwartete Segment analysiert wurde.

Du kannst die Ergebnisse jeder Iteration unabhängig sehen. Wenn eine bestimmte Iteration fehlschlägt, kannst du nur diese Iteration von der Job-Laufseite erneut ausführen, ohne den gesamten Job erneut auszuführen.

Erweitern des Musters

Um ein Segment zur Analyse hinzuzufügen, fügen Sie eine Zeile in die Kontrolltabelle ein:

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

Der nächste Job-Run beinhaltet das neue Segment, ohne Änderungen an der Job-Konfiguration oder Notebook-Bearbeitungen.

Dieses gleiche Muster funktioniert für jeden Fall, in dem Daten die Iteration steuern sollen:

• Kundenspezifische Verarbeitung: Eine Zeile pro Kunden-ID. Die verschachtelte Aufgabe wendet kundenspezifische Transformationen an oder liefert an kundenspezifische Ziele.
• Tabellenimport: Eine Zeile je Name der Quelltabelle. Die verschachtelte Aufgabe liest jede Tabelle ein.
• Backfill-Verarbeitung: Eine Zeile pro Datumspartition. Die verschachtelte Aufgabe verarbeitet historische Daten für jene Partition erneut.
• Feature-Flag-gesteuerte Ausführung: Eine Zeile pro aktiviertem Feature oder Experiment. Die verschachtelte Aufgabe aktiviert die entsprechende Logik.

Um die Verarbeitung einer Zeile zu stoppen, ohne sie zu löschen, fügen Sie eine eigene Spalte in die Steuerungstabelle (wie eine active-Flagge) hinzu und filtern Sie danach in der SQL-Nachschlage-Aufgabe. Dies ist eine gewöhnliche Spalte, die Sie definieren und befüllen; die For each-Aufgabe hat kein eingebautes Konzept davon. Füge zuerst die Spalte hinzu und setze dann die vorhandenen Zeilen auf TRUE:

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

Filtern Sie dann in der read_segments Abfrage darauf, sodass nur aktive Zeilen die Iteration steuern:

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

Weitere Ressourcen

• Verwenden einer Aufgabe zum Ausführen einer For each anderen Aufgabe in einer Schleife: Vollständige Referenz zum Konfigurieren von For each Aufgaben, einschließlich Parametertypen und Parallelitätsoptionen
• Verwenden Sie eine Nachschlagetabelle für große Parameterarrays in einer For each-Aufgabe: So behandeln Sie große Parameterarrays, die den Aufgabenwertgrenzwert von 48 KB überschreiten
• Zugreifen auf Parameterwerte aus einer Aufgabe: Alle Methoden für den Zugriff auf Parameterwerte in Notizbüchern, Python Skripts und SQL-Aufgaben
• Wanderbricks-Datensatz: Der Beispieldatensatz, der in diesem Tutorial verwendet wird