Verwende eine Steuertabelle, um einen For each Auftrag zu steuern

Wenn Sie Daten aus vielen Quellen aufnehmen, bedeutet das Hartcodieren der Liste im Einzelvorgang, dass der Code aktualisiert und bei jeder Änderung der Liste neu bereitgestellt werden muss. Verwenden Sie Metadaten , um dies zu beheben, indem Sie die Liste der Quellen in einer Tabelle speichern, die zur Laufzeit gelesen wird. Fügen Sie eine Quelle als neue Zeile hinzu und der nächste Auftrag übernimmt sie, ohne den Auftrag selbst zu ändern.

In diesem Lernprogramm erfahren Sie, wie Sie mit diesem Ansatz einen Auftrag erstellen. Eine SQL-Aufgabe liest die Steuerelementtabelle und eine For each Aufgabe durchläuft jede Zeile parallel.

So funktioniert es

Das Muster verwendet drei Aufgabentypen, die aufeinander verkabelt sind:

Aufgabe Typ Was es tut
read_markets SQL Fragt eine Konfigurationstabelle ab und erfasst das Ergebnis als Zeilenarray.
process_markets Für jede Iteriert über {{tasks.read_markets.output.rows}}, wobei die geschachtelte Aufgabe pro Zeile einmal ausgeführt wird
run_market_analysis_iteration Notebook oder SQL (verschachtelt in „For each“) Wird einmal pro Zeile ausgeführt, wobei Zeilenwerte verwendet werden, die als Parameter übergeben werden, um Ihre Geschäftslogik auszuführen.

Die Ausgabe des SQL-Tasks – ein JSON-Array von Zeilenobjekten – fließt direkt mithilfe des dynamischen Wertverweises For each in das des {{tasks.read_markets.output.rows}}-Tasks. Der For each Vorgang übergibt dann jede Zeile als Parameter an den geschachtelten Vorgang, verfügbar als {{input.market}} und {{input.currency}}.

Voraussetzungen

  • Ein Azure Databricks Arbeitsbereich mit der Berechtigung zum Erstellen von Aufträgen und Notizbüchern
  • Berechtigung zum Erstellen von Tabellen im Unity-Katalog
  • Ein Unity-Katalogschema, in dem Sie die Konfigurationstabelle erstellen können (z. B config. )
  • Ein SQL-Warehouse zum Ausführen der SQL-Aufgaben

Schritt 1: Erstellen der Konfigurationstabelle

Die Konfigurationstabelle ist die maßgebliche Quelle für die Werteliste, die Ihr Einzelvorgang verarbeitet. Wenn Sie Arbeit hinzufügen oder entfernen müssen, aktualisieren Sie diese Tabelle, nicht den Auftrag.

Führen Sie die folgende SQL-Datei aus, um eine markets Tabelle in Ihrem config Schema zu erstellen:

CREATE OR REPLACE TABLE config.markets AS
SELECT * FROM VALUES
  ('NL', 'EUR'),
  ('UK', 'GBP'),
  ('US', 'USD')
AS t(market, currency);

Verwenden Sie ein Azure Databricks-Notizbuch, den SQL-Editor oder eine sql-Aufgabe, um diese Anweisung auszuführen. Nach diesem Schritt enthält config.markets drei Zeilen, eine für jeden Markt, jeweils mit dessen Währungscode.

Schritt 2: Schreiben des Verarbeitungscodes

Die verschachtelte Aufgabe innerhalb der For each Aufgabe wird einmal pro Zeile ausgeführt. Wählen Sie je nach Geschäftslogik eine Notizbuchaufgabe oder eine SQL-Aufgabe aus.

Notebook-Aufgabe

Erstellen Sie ein neues Notizbuch unter einem Pfad wie /Workspace/Users/<username>/process_market. Dieses Notizbuch wird einmal pro Iteration der For each Aufgabe ausgeführt und erhält jedes Mal einen anderen Marktwert.

Fügen Sie dem Notizbuch den folgenden Code hinzu:

# Set default values for testing the notebook outside of a job.
# When the notebook runs inside a For each task, the job overrides these defaults.
dbutils.widgets.text("market", "NL", "Market")
dbutils.widgets.text("currency", "EUR", "Currency")

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

print(f"Processing market: {market} ({currency})")

# Your business logic goes here. For example:
df = spark.table("sales.transactions").filter(
    f"market = '{market}' AND currency_code = '{currency}'"
)
display(df)

Die dbutils.widgets.text() Aufrufe legen Standardwerte fest, damit Sie das Notizbuch direkt in Ihrem Arbeitsbereich ausführen können, ohne es mit einem Auftrag zu verbinden. Wenn das Notizbuch als geschachtelte Aufgabe innerhalb einer For each Aufgabe ausgeführt wird, überschreibt der Auftrag die Standardwerte mit den tatsächlichen Parameterwerten für diese Iteration.

Note

Rufen Sie dbutils.widgets.text() vor dbutils.widgets.get() an. Wenn get vor text aufgerufen wird, löst das Notebook einen InputWidgetNotDefined-Fehler aus, wenn Sie es außerhalb eines Einzelvorgangs ausführen.

Mithilfe von Standardeinstellungen können Sie das Notizbuch außerhalb eines Auftrags testen, aber beachten Sie: Wenn die For each Aufgabe falsch konfiguriert ist und keine Parameter übergibt, verwendet das Notizbuch die Standardwerte und ist im Hintergrund erfolgreich, anstatt fehlzuschlagen, wodurch die Fehlkonfiguration schwieriger zu erkennen ist.

SQL-Aufgabe

SQL-Aufgaben unterstützen benannte Parameter mithilfe der :param_name Syntax. Verweisen Sie :market und :currency in Ihrer Abfrage überall dort, wo Sie die Iterationswerte verwenden möchten:

SELECT *
FROM sales.transactions
WHERE market = :market
  AND currency_code = :currency

Sie konfigurieren diese Abfrage direkt im Aufgaben-Editor in Schritt 5. Die For each Aufgabe übergibt die Werte der aktuellen Iteration an die zur Laufzeit benannten :market und :currency Parameter. Im Gegensatz zu Notizbuchaufgaben unterstützen benannte SQL-Parameter keine Standardwerte. Wenn ein Parameter nicht übergeben wird, schlägt die Abfrage mit einem Fehler bei der Parameterauflösung fehl. Verwenden Sie stattdessen eine Notizbuch-Task, um Parameter zu validieren oder Standardwerte festzulegen, bevor die Abfrage ausgeführt wird.

Schritt 3: Erstellen des Auftrags

Wählen Sie in Ihrem Azure Databricks Arbeitsbereich auf der Randleiste das Plussymbol aus.Neu>Symbol Auftrag. Geben Sie dem Auftrag einen beschreibenden Namen, z. B Market Analysis. .

Schritt 4: Konfigurieren der SQL-Nachschlageaufgabe

Die SQL-Aufgabe führt Ihre Konfigurationsabfrage aus und stellt die Ausgabe für nachgeschaltete Aufgaben zur Verfügung.

  1. Klicken Sie im Auftrags-Editor auf das Plussymbol.Aufgabe hinzufügen.

  2. Legen Sie den Aufgabennamen auf read_markets.

  3. Setzen Sie Type auf SQL.

  4. Geben Sie im SQL-Feld die folgende Abfrage ein:

    SELECT market, currency FROM config.markets

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

  6. Klicken Sie auf Aufgabe speichern.

Wenn diese Aufgabe ausgeführt wird, führt Azure Databricks die Abfrage aus und erfasst das Ergebnis als JSON-Array in tasks.read_markets.output.rows. Die SQL-Aufgabenausgabe wird immer als JSON-Array zurückgegeben– es ist keine zusätzliche Konfiguration erforderlich. Die generische Form dieses Verweises ist tasks.<task-name>.output.rows, wo <task-name> mit dem Aufgabenschlüssel übereinstimmt, den Sie im Auftrags-Editor festgelegt haben. Die Ausgabe sieht wie folgt aus:

[
  { "market": "NL", "currency": "EUR" },
  { "market": "UK", "currency": "GBP" },
  { "market": "US", "currency": "USD" }
]

Schritt 5: Konfigurieren der For each Aufgabe

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

  1. Klicken Sie auf das Plus-Symbol.Aufgabe hinzufügen und stellen Sie Abhängig von auf read_markets ein.

  2. Legen Sie den Aufgabennamen auf process_markets.

  3. Legen Sie Typ auf Jeweils fest.

  4. Geben Sie im Feld "Eingaben " Folgendes ein:

    {{tasks.read_markets.output.rows}}

    Dadurch wird auf das Zeilenarray verwiesen, das von der SQL-Aufgabe erfasst wird.

  5. Legen Sie die Parallelität auf 2 fest, um zwei Iterationen parallel auszuführen. Erhöhen Sie diesen Wert, wenn Ihre geschachtelte Aufgabe höhere Parallelität unterstützt.

  6. Gehen Sie auf Aufgabe zum Ausführen einer Schleife hinzufügen und konfigurieren Sie die geschachtelte Aufgabe basierend auf dem in Schritt 2 ausgewählten Typ.

Notebook-Aufgabe

  1. Legen Sie den Aufgabennamen auf run_market_analysis_iteration.

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

  3. Legen Sie "Pfad " auf den Pfad des Notizbuchs fest, das Sie in Schritt 2 erstellt haben.

  4. Klicken Sie auf "Parameter", und klicken Sie dann auf "Hinzufügen" , um die folgenden Parameter hinzuzufügen:

    • Schlüssel: , market: {{input.market}}
    • Schlüssel: , currency: {{input.currency}}

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

  5. Klicken Sie auf Aufgabe speichern.

SQL-Aufgabe

  1. Legen Sie den Aufgabennamen auf run_market_analysis_iteration.

  2. Setzen Sie Type auf SQL.

  3. Geben Sie im SQL-Feld Ihre Abfrage mit den benannten Parametern ein, z. B.:

    SELECT *
FROM sales.transactions
WHERE market = :market
  AND currency_code = :currency

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

  5. Klicken Sie auf "Parameter", und klicken Sie dann auf "Hinzufügen" , um die folgenden Parameter hinzuzufügen:

    • Schlüssel: , market: {{input.market}}
    • Schlüssel: , currency: {{input.currency}}

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

  6. Klicken Sie auf Aufgabe speichern.

Der gerichtete azyklische Graph für Ihren Einzelvorgang zeigt nun, wie read_markets in process_markets fließt, wobei die geschachtelte Aufgabe innerhalb des For each-Knotens sichtbar ist.

Schritt 6: Ausführen des Auftrags und Überprüfen

  1. Klicken Sie auf "Jetzt ausführen" , um den Auftrag auszulösen.
  2. Klicken Sie auf der Auftragsausführungsseite auf den process_markets Knoten, um die For each Aufgabe zu erweitern.
  3. Die Auftragsausführungsseite zeigt eine Tabelle mit Iterationen, eine Zeile pro Marktwert, wobei jeweils der Status, die Startzeit und die Dauer angezeigt werden.
  4. Klicken Sie auf eine beliebige Iterationszeile, um das Ergebnis der Aufgabenausführung zu öffnen und zu bestätigen, dass das Ergebnis den richtigen Marktwert erhalten hat.

Wenn eine bestimmte Iteration fehlschlägt, führen Sie nur diese Iteration von der Auftragsausführungsseite aus, ohne den gesamten Auftrag erneut auszuführen.

Erweitern des Musters

Um einen neuen Markt hinzuzufügen, fügen Sie eine Zeile in die Konfigurationstabelle ein:

INSERT INTO config.markets VALUES ('DE', 'EUR');

Der nächste Auftrag wird automatisch in Deutschland ausgeführt, ohne dass Auftragskonfigurationsänderungen oder Notizbuchbearbeitungen erforderlich sind.

Dieses Muster funktioniert für jeden Anwendungsfall, in dem Daten die Iteration fördern sollen:

  • Kundenspezifische Verarbeitung: Eine Zeile pro Kunden-ID. Das Notebook wendet kundenspezifische Transformationen an oder liefert an kundenspezifische Zielorte.
  • Tabellenimport: Eine Zeile je Name der Quelltabelle. Das Notizbuch liest und nimmt jede Tabelle ein.
  • Backfill-Verarbeitung: Eine Zeile pro Datumspartition. Das Notizbuch verarbeitet historische Daten für diese Partition erneut.
  • Feature-Flag-gesteuerte Ausführung: Eine Zeile pro aktiviertem Feature oder Experiment. Das Notizbuch aktiviert die entsprechende Logik.

Wenn Sie ein Element aus der Verarbeitung entfernen möchten, löschen Sie die Zeile, oder fügen Sie in der SQL-Abfrage eine active Kennzeichnungsspalte und einen Filter hinzu:

SELECT market, currency FROM config.markets WHERE active = TRUE

Weitere Ressourcen

  • Last updated on