Ordnen der Ausführung des Pipeline-Flows mit depends_on

Important

Dieses Feature befindet sich in der Public Preview.

Um zu verwenden depends_on, konfigurieren Sie Ihre Pipeline so, dass sie den PREVIEW-Kanal der Lakeflow-Pipelines verwendet. Siehe channel in Pipelinekonfigurationen.

Standardmäßig plant eine Pipeline Flows nach ihren Datenabhängigkeiten: Wenn ein Flow die Tabelle liest, die ein anderer Flow schreibt, folgt der Reader dem Schreiber. Durch die Flussordnung kann ein Fluss auf einen anderen Fluss warten, aus dem er nicht liest, indem die Abhängigkeit explizit mit depends_ondeklariert wird.

depends_on="other_flow" bedeutet, dass dieser Fluss erst beginnt, nachdem other_flow erfolgreich abgeschlossen wurde. Es ist ein Scheduling-Vorteil: Er steuert, wann ein Flow beginnt, nicht wie er läuft. Das Deklarieren depends_on ändert weder den Trigger, den Modus noch die Frage, ob es sich um einen einmaligen Flow handelt. Du deklarierst diese immer noch im Flow selbst, zum Beispiel mit once=True.

Wann die Flow-Reihenfolge verwendet werden sollte

Der Hauptanwendungsfall besteht darin, historische Daten zu entnehmen, bevor auf eine Live-Quelle umgestellt wird, während der Streaming-Status erhalten bleibt, etwa indem eine Tabelle von einem Batch-Backfill auf eine Live-Apache-Kafka-Quelle übertragen wird.

Das gleichzeitige Lesen beider Quellen funktioniert nicht gut: Das begrenzte Backfill hält die Watermark zurück, was die Zustandsbereinigung und die Fensterergebnisse verzögert. Das lässt sich vermeiden, indem zuerst der Backfill beendet und dann der Live-Stream gestartet wird. Die Flowordnung sequenziert die beiden.

Die Flussordnung unterstützt auch andere Muster:

  • Mehrere Backfills strikt in dieselbe Tabelle ordnen.
  • Pipelines mit mehreren Phasen, wie zum Beispiel ein anfänglicher Ladevorgang, dann ein Nachziehen, dann ein Echtzeitdatenstrom.
  • Sequenzierungsflüsse, die in verschiedene Tabellen schreiben, aber in einer festgelegten Reihenfolge ausgeführt werden müssen.

Flows mit depends_on ordnen

depends_on ist in der Python-Pipelines-API für die @dp.append_flow- und @dp.update_flow-Dekoratoren verfügbar. Es akzeptiert einen einzelnen Flussnamen oder eine Liste von Flussnamen. Bei einer Liste muss jeder benannte Flow abgeschlossen sein, bevor der abhängige Flow startet.

Das folgende Beispiel schreibt einen einmaligen Backfill-Vorgang in die Streaming-Tabelle events und startet erst dann einen Live-Kafka-Stream in dieselbe Tabelle, nachdem der Backfill abgeschlossen ist:

from pyspark import pipelines as dp

dp.create_streaming_table(name="events")

# Drain the historical backfill first.
@dp.append_flow(target="events", once=True, name="events_backfill")
def events_backfill():
    return spark.read.table("historical_events")

# Start the live stream only after the backfill completes.
@dp.append_flow(target="events", name="events_live", depends_on="events_backfill")
def events_live():
    return (
        spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", "<server>:<port>")
        .option("subscribe", "events")
        .load()
    )

Um auf mehr als einen Vorgänger zu warten, reichen Sie eine Liste weiter. Das folgende Beispiel führt zwei Backfills parallel aus und startet den Live-Stream erst, nachdem beide abgeschlossen sind:

@dp.append_flow(target="events", once=True, name="backfill_2024")
def backfill_2024():
    return spark.read.table("events_2024")

@dp.append_flow(target="events", once=True, name="backfill_2025")
def backfill_2025():
    return spark.read.table("events_2025")

@dp.append_flow(
    target="events",
    name="events_live",
    depends_on=["backfill_2024", "backfill_2025"],
)
def events_live():
    return spark.readStream.format("kafka").option("subscribe", "events").load()

Um Backfills nacheinander statt parallel zu ordnen, verketten Sie depends_on über sie:

@dp.append_flow(target="events", once=True, name="backfill_2024")
def backfill_2024():
    return spark.read.table("events_2024")

@dp.append_flow(
    target="events", once=True, name="backfill_2025", depends_on="backfill_2024"
)
def backfill_2025():
    return spark.read.table("events_2025")

Anforderungen und Verhalten

Für die Flussordnung gelten folgende Regeln:

  • Die Ablaufreihenfolge funktioniert nur innerhalb einer Pipeline. Ohne den Pipeline-Scheduler kann die Reihenfolge nicht eingehalten werden.
  • Ein Vorgänger muss in eine Tabelle oder Senke schreiben, nicht in eine Ansicht. Ein Flow, der in eine Ansicht schreibt, erreicht nie einen terminalen Zustand, sodass ein Flow, der danach angeordnet wird, nie starten würde. Streaming Table Targets und foreachBatch Sinks sind beide gültige Vorgänger.
  • Destinationsübergreifende Bestellungen sind zulässig. Ein Flow kann von einem Flow abhängen, der in eine andere Tabelle schreibt.
  • Unbekannte Flussnamen und -zyklen werden bei der Validierung erkannt, bevor die Pipeline läuft.

Was kann ein Vorgänger in getriggerten und kontinuierlichen Pipelines sein?

Die Arten von Flows, die als Vorgänger fungieren können, hängen vom Ausführungsmodus der Pipeline ab:

  • Auslösende Pipelines: Jeder Flow kann als Vorgänger fungieren. Jeder Fluss in einem ausgelösten Lauf erreicht einen Endzustand, sodass die Reihenfolge innerhalb jedes Laufs gilt.
  • Kontinuierliche Pipelines: Ein Vorgänger muss ein einmaliger (once)-Fluss sein, der einen Endzustand erreicht. Ein kontinuierlich laufender Fluss endet nie, sodass ein danach angeordneter Fluss nie starten würde, und die Pipeline lehnt ihn bei der Validierung ab.

Da ein foreachBatch Flow immer ein Streaming-Sink ist und kein einmaliger Flow sein kann, kann er nur in einer ausgelösten Pipeline als Vorgänger fungieren. In einer kontinuierlichen Pipeline kann ein foreachBatch Flow auf einen Vorgänger once warten, aber er kann selbst kein Vorgänger sein.

Betriebsverhalten

Der Fertigstellungszustand eines Vorgängers once ist dauerhaft, sodass er Neustarts und Pipeline-Updates übersteht:

  • Neustart: bereits geleerte Flows bleiben geleert und werden übersprungen. Die Pipeline wird beim ersten Fluss wieder aufgenommen, der noch nicht abgeschlossen ist. Nachdem die Ausführung den Live-Flow erreicht hat, setzen spätere Neustarts nur den Live-Flow fort.
  • Volle Aktualisierung: löscht den Abschlusszustand der Kette und führt sie von Anfang an in der richtigen Reihenfolge erneut aus.
  • Zurücksetzen eines Checkpoints für einen einzelnen Flow: Beim Zurücksetzen des Live-Flows wird nur der Live-Flow erneut ausgeführt. Vorgelagerte once-Backfills bleiben geleert und werden nicht erneut ausgeführt. Dies ist die übliche Methode, eine Live-Abfrage wiederherzustellen, ohne die Historie erneut zu entladen.

Weitere Ressourcen

  • Zuletzt aktualisiert um