Uitvoeringsvolgorde van pijplijnstromen met depends_on bepalen

Om depends_on te gebruiken, configureer je pijplijn om het PREVIEW-kanaal te gebruiken. Zie channel in Pijplijnconfiguraties.

Standaard plant een pijplijn stromen op basis van hun gegevensafhankelijkheden: als een stroom de tabel leest die een andere stroom schrijft, loopt de lezer achter de schrijver aan. Flow-ordening laat een flow wachten op een andere flow waarvan hij niet leest, door de afhankelijkheid expliciet met depends_onte declareren.

depends_on="other_flow" betekent dat deze flow pas begint nadat other_flow met succes is voltooid. Het is een planningsvoordeel: het bepaalt wanneer een flow begint, niet hoe het verloopt. Declareren depends_on verandert niet de trigger, modus of of het een eenmalige flow is. Je declareert die nog steeds op de flow zelf, bijvoorbeeld met once=True.

Wanneer flow-ordening gebruiken

Het primaire gebruiksscenario is het aftappen van historische data voordat wordt overgeschakeld naar een live bron, terwijl de streamingstatus behouden blijft, zoals het migreren van een tabel van een batch backfill naar een live Apache Kafka-bron.

Het gelijktijdig lezen van beide bronnen werkt niet goed: de begrensde backfill houdt de watermark tegen, waardoor het opschonen van status en de vensterresultaten worden vertraagd. Eerst de backfill leegpompen en daarna de livestream starten, voorkomt dit. Flow-volgorde rangschikt de twee.

Flowvolgorde ondersteunt ook andere patronen:

• Strikt meerdere backfills in dezelfde tabel bestellen.
• Gefaseerde pipelines, zoals een eerste gegevenslading, daarna een inhaalslag en vervolgens een livefeed.
• Sequencingflows die naar verschillende tabellen schrijven, maar in een vaste volgorde moeten draaien.

Orde stroomt mee met depends_on

depends_onis beschikbaar op de @dp.append_flow en @dp.update_flow decorators in de Python pipelines API. Het accepteert één enkele stroomnaam of een lijst met stroomnamen. Bij een lijst moet elke benoemde flow voltooid zijn voordat de afhankelijke flow begint.

Het volgende voorbeeld voert een eenmalige backfill af in de events streaming-tabel en start daarna pas na afloop van de backfill een live Kafka-stream in dezelfde tabel:

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

Om op meer dan één voorganger te wachten, geef een lijst door. Het volgende voorbeeld voert twee backfills parallel uit en start de livestream pas nadat beide zijn voltooid:

@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()

Om backfills achter elkaar uit te voeren in plaats van parallel, koppel je depends_on ertussen aan elkaar:

@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")

foreachBatch-stromen ordenen

Gebruik depends_on op de append flow die gericht is op een foreachBatch sink, niet op de @dp.foreach_batch_sink decorator. Een foreachBatch flow kan geen eenmalige (once) flow zijn. Zie ForEachBatch gebruiken om in pijplijnen naar willekeurige datasinks te schrijven voor overwegingen met betrekking tot sinkgedrag en nieuwe pogingen.

Sequence foreachBatch-flows in een geactiveerde pijplijn

In een getriggerde pijplijn kan een foreachBatch flow zowel een voorganger als een opvolger zijn. Het volgende voorbeeld voert een backfill uit op een streamingtabel, exporteert de rijen via een foreachBatch sink en verwerkt vervolgens dezelfde rijen via een tweede foreachBatch sink.

Beide sinkstromen luiden ordered_events. De depends_on="export_events" afhankelijkheid plaatst de auditstroom na de exportflow, ook al leest de auditflow de exporttabel niet. Vervang <catalog> en <schema> door de catalogus en het schema voor je bron- en bestemmingstabellen.

import dlt
from pyspark import pipelines as dp

dp.create_streaming_table(name="ordered_events")


@dp.append_flow(target="ordered_events", once=True, name="events_backfill")
def events_backfill():
    return spark.read.table("<catalog>.<schema>.historical_events")


@dp.foreach_batch_sink(name="events_export_sink")
def export_batch(batch_df, batch_id):
    batch_df.write.format("delta").mode("append").saveAsTable(
        "<catalog>.<schema>.exported_events"
    )


@dp.append_flow(
    target="events_export_sink", name="export_events", depends_on="events_backfill"
)
def export_events():
    return dlt.read_stream("ordered_events")


@dp.foreach_batch_sink(name="events_audit_sink")
def audit_batch(batch_df, batch_id):
    batch_df.write.format("delta").mode("append").saveAsTable(
        "<catalog>.<schema>.audited_events"
    )


@dp.append_flow(
    target="events_audit_sink", name="audit_events", depends_on="export_events"
)
def audit_events():
    return dlt.read_stream("ordered_events")

Voer een foreachBatch-flow uit na een backfill in een continue pipeline

In een continue pipeline kan een foreachBatch flow een opvolger zijn, maar geen voorganger. De voorganger moet een eenmalige (once) flow zijn. Het volgende voorbeeld verwerkt de backfill-rijen via een foreachBatch sink nadat de backfill is voltooid:

import dlt
from pyspark import pipelines as dp

dp.create_streaming_table(name="backfilled_events")


@dp.append_flow(target="backfilled_events", once=True, name="events_backfill")
def events_backfill():
    return spark.read.table("<catalog>.<schema>.historical_events")


@dp.foreach_batch_sink(name="events_export_sink")
def export_batch(batch_df, batch_id):
    batch_df.write.format("delta").mode("append").saveAsTable(
        "<catalog>.<schema>.continuous_exported_events"
    )


@dp.append_flow(
    target="events_export_sink", name="export_events", depends_on="events_backfill"
)
def export_events():
    return dlt.read_stream("backfilled_events")

Voeg geen extra flow met depends_on="export_events" toe aan dit continue voorbeeld. De exportflow is geen eenmalige flow, dus de pijplijn verwerpt deze als voorgangerstaak met DEPENDS_ON_PREDECESSOR_MUST_BE_ONCE.

Vereisten en gedrag

De volgende regels zijn van toepassing op de volgorde van stromen:

• Stroomvolgorde functioneert alleen binnen een pijplijn. Zonder de pipeline-scheduler kan de volgorde niet worden nageleefd.
• Een voorganger moet schrijven op een tafel of gootsteen, niet op een weergave. Een flow die naar een view schrijft bereikt nooit een terminale toestand, dus een flow die daarna wordt besteld, zou nooit starten. Doeltabellen voor streamingtabellen zijn geldige voorgaande objecten. Een foreachBatch sinkflow kan een voorganger zijn in een getriggerde pijplijn, maar niet in een continue pijplijn.
• Bestellen over verschillende bestemmingen heen is toegestaan. Een flow kan afhangen van een flow die naar een andere tabel schrijft.
• Onbekende stromennamen en cycli worden bij validatie vastgelegd, voordat de pijplijn draait.

Wat kan een voorganger zijn in getriggerde en continue pijplijnen

De soorten flows die als voorganger kunnen fungeren, hangen af van de uitvoeringsmodus van de pipeline:

• Getriggerde pijplijnen: elke flow kan een voorganger zijn. Elke stroom in een getriggerde run bereikt een terminale toestand, dus de volgorde geldt binnen elke run.
• Doorlopende pijplijnen: een voorganger moet een eenmalige (once) stroom zijn die een eindtoestand bereikt. Een stroom die continu draait eindigt nooit, dus een stroom die daarna wordt besteld zou nooit starten, en de pijplijn wijst het af bij validatie.

Omdat een foreachBatch flow altijd een streaming sink is en geen eenmalige flow kan zijn, kan het alleen als voorganger optreden in een getriggerde pijplijn. In een continue pijplijn kan een foreachBatch flow wachten op een once voorganger, maar hij kan zelf geen voorganger zijn.

Operationeel gedrag

De voltooiingsstatus voor een once voorganger is duurzaam, dus deze overleeft herstarts en updates van de pijplijn:

• Herstart: al afgevoerde stromen blijven leeg en worden overgeslagen. De pijpleiding wordt hervat bij de eerste stroming die nog niet is voltooid. Nadat de uitvoering de live flow bereikt, hervatten latere herstarts alleen de live flow.
• Volledige verversing: verwijdert de voltooiingsstatus voor de keten en voert deze opnieuw uit vanaf het begin, in volgorde.
• Checkpoint reset voor één enkele flow: het resetten van de live flow speelt alleen de live flow af. Upstream-oncebackfills blijven leeggemaakt en worden niet opnieuw uitgevoerd. Dit is de gebruikelijke manier om een live query te herstellen zonder de geschiedenis opnieuw te belasten.

Aanvullende bronnen

• Gegevens incrementeel laden en verwerken met Lakeflow-pijplijnstromen
• Het aanvullen van historische gegevens met datapijplijnen
• Gebruik ForEachBatch om naar willekeurige datasinks in pijplijnen te schrijven
• append_flow
• update_flow