Użyj sinka ForEach do zapisywania każdego wiersza w pipeline’ach

Użyj sinka ForEach do zapisu danych streamingowych do miejsc, które nie mają wbudowanego sinka, takich jak zewnętrzne API REST czy magazyn key-value.

Sink ForEach przetwarza strumień po jednym wierszu na raz. Dla każdego wiersza potok wywołuje niestandardową funkcję Python lub obiekt writer, podobnie jak sink foreach w Apache Spark Structured Streaming.

Kiedy używać ujścia ForEach

Użyj elementu sink ForEach, gdy musisz zapisać każdy rekord osobno w systemie docelowym bez wbudowanego formatu sinka, na przykład wywołując zewnętrzne API lub aktualizując magazyn klucz-wartość dla każdego rekordu. Zamiast tego używaj ForEachBatch ujścia, gdy twoja logika działa na całej mikropartii, na przykład gdy scala dane z tabelą Delta Lake lub zapisuje je do kilku miejsc docelowych za pomocą operacji DataFrame.

Ta tabela porównuje oba zlewy:

Aby uzyskać informacje o wbudowanych sinkach, zobacz Sinks in Lakeflow pipelines. W odniesieniu do @dp.foreach_sink() dokumentacji referencyjnej API Pythona zobacz foreach_sink.

Zdefiniuj ujście ForEach

Możesz zdefiniować ForEach ujście na podstawie funkcji lub obiektu zapisu.

Aby zdefiniować ujście na podstawie funkcji, opatrz funkcję adnotacją @dp.foreach_sink(). Pipeline wywołuje funkcję dla każdego wiersza. Jeśli pominiesz name, sink używa nazwy funkcji:

from pyspark import pipelines as dp

@dp.foreach_sink()
def my_row_sink(row):
    # Custom per-row logic here.
    ...

Aby zdefiniować sink z obiektu writer, przekaż obiekt za pomocą argumentu słowa kluczowego writer. Autor musi zaimplementować process(row) i może zaimplementować open(partition_id, epoch_id) oraz close(error). Używaj writera, gdy chcesz otworzyć jedno połączenie dla każdej partycji i użyć go ponownie dla każdego wiersza. Jeśli pominiesz name, sink używa nazwy klasy writera, zapisanej małymi literami:

from pyspark import pipelines as dp

class MyRowWriter:
    def open(self, partition_id, epoch_id):
        # Open a connection for this partition.
        return True

    def process(self, row):
        # Write the row by using the connection.
        ...

    def close(self, error):
        # Close the connection.
        ...

dp.foreach_sink(name="my_row_sink", writer=MyRowWriter())

Szczegóły dotyczące każdej metody znajdziesz w metodach Writer.

Wpisy dziennika zdarzeń

Gdy definiujesz sink ForEach, dziennik zdarzeń pipeline rejestruje zdarzenie SinkDefinition z "format": "foreach" dla sinka.

Pisz do ujścia ForEach

Możesz zapisywać do sink ForEach z przepływu append lub update.

Zapisuj w trybie dopisywania

Aby zapisać do ujścia ForEach z przepływu dołączania, odwołaj się do nazwy ujścia jako target:

@dp.append_flow(target="my_row_sink")
def my_flow():
    return spark.readStream.table("my_catalog.my_schema.source_table")

Możesz zdefiniować wiele przepływów dołączania, które są kierowane do tego samego ForEach ujścia. Każdy przepływ ma swój własny punkt kontrolny.

Pisz w przepływie aktualizacji

Aby zapisać do sinku ForEach z przepływu aktualizacji, użyj dekoratora @dp.update_flow:

@dp.update_flow(target="my_row_sink")
def my_stateful_flow():
    return spark.readStream.table("my_catalog.my_schema.events").groupBy("user_id").count()

Przepływ aktualizacji może wysłać ten sam klucz do ujścia ponownie w późniejszej mikropartii z nową wartością. Aby zastąpić starą wartość, wykonaj upsert każdego wiersza w miejscu docelowym za pomocą klucza zamiast go wstawiać. W poprzednim przykładzie użyj user_id jako klucza, tak aby każde nowe zliczenie dla użytkownika zastępowało poprzednie.

W przypadku przetwarzania o niskich opóźnieniach proces aktualizacji, który zapisuje do ujścia ForEach, może działać w trybie czasu rzeczywistego. Zobacz Korzystanie z trybu czasu rzeczywistego w potokach Lakeflow.

Importuj biblioteki i zależności na egzekutorach

Twoja funkcja lub writer działa na executors, a nie na driverze. Postępuj zgodnie z następującymi rozwiązaniami:

• Dodaj pakiety, których potrzebujesz w kodzie, do środowiska pipeline. Pakiety w środowisku pipeline są dostępne dla twojego kodu na wykonawcach. Zobacz Zarządzanie zależnościami języka Python dla potoków.
• Importuj biblioteki, których używasz w swoim kodzie, wewnątrz funkcji lub metod writera, nie tylko na górze pliku źródłowego pipeline.
• Nie polegaj na obiektach tworzonych w sterowniku, takich jak klient czy połączenie, które otwierasz na początku pliku. Utwórz je wewnątrz funkcji lub w metodzie open modułu zapisującego.

Obsługa błędów i zduplikowanych wierszy

Jeśli Twój kod wywoła wyjątek, przepływ się nie udaje, a dziennik zdarzeń potoku rejestruje zdarzenie błędu, które zawiera komunikat wyjątku.

Gdy mikropartia jest ponawiana po awarii, ujście może ponownie otrzymać kilka wierszy. Spraw, aby operacje zapisu dla każdego wiersza były idempotentne, na przykład używając unikalnego klucza z wiersza do upsertowania do miejsca docelowego. Zobacz Gwarancje przetwarzania w potokach Lakeflow.

Jeśli użyjesz obiektu zapisującego, open otrzymuje partition_id i epoch_id. Razem te ID identyfikują wiersze jednej partycji w jednej mikropartii, więc możesz ich użyć do deduplikacji wierszy. Funkcja nie otrzymuje tych identyfikatorów.

Identyfikatory są unikalne tylko w obrębie punktu kontrolnego jednego przepływu. Ta sama para może się powtórzyć w następujących przypadkach:

• Kilka przepływów zapisuje do tego samego ujścia.
• Pełne odświeżenie resetuje punkty kontrolne.

Aby deduplikować za pomocą tych ID, zapisz do sinku z pojedynczego przepływu i wyczyść stan deduplikacji podczas pełnego odświeżania.

Pełne odświeżanie

Po pełnym odświeżeniu pipeline resetuje punkt kontrolny każdego przepływu zapisującego do ForEach sink. Pipeline nie śledzi ani nie czyści danych, które zapisuje sink, więc to ty zarządzasz retencją i czyszczeniem dla swojego miejsca docelowego. Jeśli potrzebujesz czystego początku, usuń te dane samodzielnie.

Limitations

• Metryki wierszy: Potok nie raportuje liczby wierszy, które zapisuje sink ForEach .
• Wywołania na wiersz: Jeśli Twój kod wywołuje zewnętrzny system dla każdego wiersza, system zewnętrzny może ograniczyć przepustowość. Aby zmniejszyć narzut, otwieraj połączenia w metodzie open modułu zapisującego i używaj ich ponownie dla każdego wiersza lub użyj sink ForEachBatch do zapisu zbiorczego.

Examples

Te przykłady pokazują powszechne sposoby korzystania z elementu sink ForEach.

Podstawowy przykład składni

Ten przykład definiuje ForEach sink z funkcji i zapisuje do niej z przepływu append:

from pyspark import pipelines as dp

# Create a ForEach sink
@dp.foreach_sink(name="my_foreach_sink")
def handle_row(row):
    # Custom per-row logic here.
    ...

# Create source data for the example
@dp.table()
def example_source_data():
    return spark.range(5)

# Write to the sink from an append flow
@dp.append_flow(target="my_foreach_sink")
def my_flow():
    return spark.readStream.table("example_source_data")

Zapisz każdy wiersz do zewnętrznego API

W tym przykładzie używa się obiektu writer do wysyłania każdego zamówienia do zewnętrznego REST API. Autor otwiera jedną sesję HTTP dla każdej partycji. Używa identyfikatora zamówienia w URL, więc powtórzony wiersz aktualizuje ten sam rekord zamiast tworzyć duplikat:

from pyspark import pipelines as dp

class OrdersApiWriter:
    def open(self, partition_id, epoch_id):
        import requests  # Import on the executor

        self.session = requests.Session()
        return True

    def process(self, row):
        response = self.session.put(
            f"https://api.example.com/orders/{row.order_id}",
            json={"order_id": row.order_id, "status": row.status},
            timeout=10,
        )
        # Raise on HTTP errors so that the flow fails instead of dropping the row.
        response.raise_for_status()

    def close(self, error):
        self.session.close()

dp.foreach_sink(name="orders_api_sink", writer=OrdersApiWriter())

@dp.append_flow(target="orders_api_sink")
def orders_to_api():
    return spark.readStream.table("my_catalog.my_schema.orders")

Zapisywanie do jednego sinka z wielu przepływów danych

Ten przykład zapisuje dwa źródła do tego samego ForEach ujścia:

from pyspark import pipelines as dp

@dp.foreach_sink(name="events_sink")
def handle_event(row):
    # Custom per-row logic here.
    ...

@dp.append_flow(target="events_sink")
def web_events():
    return spark.readStream.table("my_catalog.my_schema.web_events")

@dp.append_flow(target="events_sink")
def mobile_events():
    return spark.readStream.table("my_catalog.my_schema.mobile_events")

Dodatkowe zasoby

• foreach_sink
• Użyj ForEachBatch do zapisu do dowolnych miejsc docelowych danych w potokach
• Użyj foreachBatch do zapisu do dowolnych ujść danych
• Gwarancje przetwarzania w rurociągach Lakeflow
• Korzystanie z trybu czasu rzeczywistego w potokach Lakeflow