Wykonanie przepływu potoku zamówień za pomocą depends_on

Aby użyć depends_on, skonfiguruj swój pipeline tak, aby korzystał z kanału PREVIEW. Zobacz channel w temacie Konfiguracje potoków.

Domyślnie potok harmonogramuje przepływy na podstawie ich zależności danych: jeśli jeden przepływ odczytuje tabelę, którą zapisuje inny przepływ, przepływ odczytujący jest uruchamiany po przepływie zapisującym. Porządkowanie przepływu pozwala przepływowi czekać na inny przepływ, z którego nie odczytuje, poprzez jawne deklarowanie zależności za pomocą .depends_on

depends_on="other_flow" oznacza, że ten przepływ zaczyna się dopiero po pomyślnym zakończeniu other_flow. To jest przewaga planowania: kontroluje, kiedy zaczyna się przepływ, a nie jak działa. Deklarowanie depends_on nie zmienia wyzwalacza, trybu ani tego, czy jest to jednorazowy przepływ. Nadal deklarujesz je bezpośrednio w przepływie, na przykład za pomocą once=True.

Kiedy stosować kolejność przepływu

Głównym przypadkiem użycia jest wyczerpanie danych historycznych przed przejściem na źródło na żywo, przy jednoczesnym zachowaniu stanu streamingowego, na przykład migracja tabeli z wsadowego uzupełnienia do źródła Apache Kafka.

Jednoczesne odczytywanie obu źródeł nie sprawdza się: ograniczony backfill wstrzymuje znacznik wodny, co opóźnia usuwanie stanu i generowanie wyników okienkowych. Najpierw opróżnienie zasypu, a potem rozpoczęcie transmisji na żywo, pozwala temu uniknąć. Porządkowanie przepływu ustala kolejność tych dwóch elementów.

Kolejność przepływu wspiera także inne wzorce:

• Ścisłe uporządkowanie kilku uzupełnień danych do tej samej tabeli.
• Potoki fazowane, takie jak początkowe obciążenie, potem nadrabianie, a następnie transmisja na żywo.
• Sekwencjonowanie przepływów zapisujących do różnych tabel, ale muszących być uruchamianymi w określonej kolejności.

Uporządkuj przepływy za pomocą depends_on

depends_on jest dostępny dla dekoratorów @dp.append_flow i @dp.update_flow w interfejsie API potoków Pythona. Akceptuje pojedynczą nazwę przepływu lub listę nazw przepływów. W przypadku listy każdy nazwany przepływ musi się zakończyć przed rozpoczęciem zależnego przepływu.

Poniższy przykład przesyła jednorazowe uzupełnienie danych do tabeli strumieniowej events, a następnie uruchamia strumień danych Kafka do tej samej tabeli dopiero po zakończeniu uzupełniania danych:

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

Aby czekać na więcej niż jednego poprzednika, przekażcie listę. Poniższy przykład uruchamia dwa uzupełnienia równolegle i rozpoczyna transmisję na żywo dopiero po zakończeniu obu tych etapów:

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

Aby uruchamiać uzupełnienia jedno po drugim zamiast równolegle, połącz je w łańcuch za pomocą depends_on:

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

Sortowanie przepływów

Użyj depends_on w przepływie dołączania kierowanym do sinka foreachBatch, a nie w dekoratorze @dp.foreach_batch_sink. Przepływ foreachBatch nie może być przepływem uruchamianym tylko raz (once). Aby poznać zachowanie ujścia i zagadnienia dotyczące ponownych prób, zobacz Używanie ForEachBatch do zapisywania do dowolnych ujść danych w potokach.

Sekwencja przepływów foreachBatch w potoku wyzwalanym

W potoku wyzwalanym foreachBatch przepływ może być zarówno poprzednikiem, jak i następcą. Poniższy przykład uzupełnia tabelę strumieniową, eksportuje jej wiersze przez foreachBatch sink, a następnie przetwarza te same wiersze przez drugi foreachBatch sink.

Oba przepływy sink mają wartość ordered_events. Zależność depends_on="export_events" sekwencjonuje przepływ audytu po przepływie eksportowym, mimo że przepływ audytu nie odczytuje tabeli eksportu. Zamień <catalog> i <schema> na katalog i schemat dla tabel źródłowych i docelowych.

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

Uruchom przepływ Flow foreachBatch po backfillu w ciągłym potoku

W ciągłym potoku przepływ foreachBatch może być następcą, ale nie poprzednikiem. Jego poprzednik musi być przepływem jednorazowym (once). Poniższy przykład przetwarza wiersze uzupełnione wstecznie przez ujście foreachBatch po zakończeniu uzupełniania wstecznego:

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

Nie dodawaj kolejnego przepływu z depends_on="export_events" do tego ciągłego przykładu. Przepływ eksportu nie jest przepływem jednorazowym, więc potok odrzuca go jako poprzednika z DEPENDS_ON_PREDECESSOR_MUST_BE_ONCE.

Wymagania i zachowanie

Następujące zasady dotyczą porządkowania przepływu:

• Kolejność w przepływie działa tylko w obrębie pipeline’u. Bez mechanizmu harmonogramowania potoku nie można zachować tej kolejności.
• Poprzednik musi pisać do tabeli lub zlewu, a nie do widoku. Przepływ, który zapisuje do widoku, nigdy nie osiąga stanu końcowego, więc przepływ następujący po nim nigdy się nie rozpocznie. Cele tabel strumieniowych są prawidłowymi poprzednikami. Przepływ ujścia foreachBatch może być poprzednikiem w potoku wyzwalanym, ale nie w ciągłym.
• Sortowanie między miejscami docelowymi jest dozwolone. Przepływ może zależeć od przepływu, który zapisuje do innej tabeli.
• Nieznane nazwy i cykle przepływów są wychwytywane podczas walidacji, zanim pipeline ruszy.

Co może być poprzednikiem w potokach uruchamianych i ciągłych

Typy przepływów, które mogą pełnić rolę poprzednika, zależą od trybu wykonywania potoku:

• Potoki wyzwalane: dowolny przepływ może pełnić rolę poprzednika. Każdy przepływ w wyzwalanym uruchomieniu osiąga stan końcowy, więc ta kolejność ma zastosowanie w ramach każdego uruchomienia.
• Potoki ciągłe: poprzednik musi być jednorazowym (once) przepływem, który osiąga stan terminalny. Przepływ działający nieprzerwanie nigdy się nie kończy, więc przepływ następujący po nim nigdy się nie rozpocznie, a potok przetwarzania odrzuca go na etapie walidacji.

Ponieważ przepływ foreachBatch jest zawsze ujściem strumieniowym i nie może być przepływem jednorazowym, może pełnić rolę poprzednika tylko w potoku wyzwalanym. W potoku ciągłym przepływ foreachBatch może czekać na poprzedni przepływ once, ale sam nie może być poprzednikiem.

Zachowanie operacyjne

Stan zakończenia dla poprzednika once jest trwały, więc przetrwa ponowne uruchomienia i aktualizacje potoku:

• Restart: już opróżnione przepływy pozostają opróżnione i są pomijane. Rurociąg wznawia się przy pierwszym przepływie, który jeszcze się nie zakończył. Gdy wykonywanie przejdzie do aktywnego przepływu, kolejne ponowne uruchomienia wznawiają tylko aktywny przepływ.
• Pełne odświeżenie: resetuje stan ukończenia łańcucha i uruchamia go ponownie od początku, w kolejności.
• Resetowanie punktu kontrolnego dla pojedynczego przepływu: resetowanie aktywnego przepływu powoduje ponowne odtworzenie tylko aktywnego przepływu. Zasypy powyżej rzeki once pozostają odwadniane i nie są ponownie uruchamiane. To typowy sposób na przywrócenie aktywnego zapytania bez ponownego opróżniania historii.

Dodatkowe zasoby

• Ładuj i przetwarzaj dane stopniowo za pomocą przepływów potoku Lakeflow
• Wypełnianie danych historycznych za pomocą potoków
• Użyj ForEachBatch do zapisu do dowolnych miejsc docelowych danych w potokach
• append_flow
• update_flow