foreach_sink

Dekorator @dp.foreach_sink() definiuje sink, ForEach który przetwarza strumień wiersz po wierszu za pomocą własnej logiki Python. Zwracasz się do tego zlewu jako target do API w flow append lub update flow , aby zapisać dane. Dla wskazówek koncepcyjnych, rozważań i przykładów zobacz: Użyj zlewu ForEach do zapisu każdego wiersza w potokach. Aby przetworzyć każdą mikropartię jako DataFrame, zobacz foreach_batch_sink.

Syntax

Aby zdefiniować ForEach zlew z funkcji, ozdob tę funkcję na:@dp.foreach_sink()

from pyspark import pipelines as dp

@dp.foreach_sink(name="<name>")
def row_handler(row):
    """
    Required:
      - `row`: a Row that contains one record of the stream.
    """
    # Your custom write logic here.
    # This function runs on the executors, so import any libraries it uses inside the function.

Aby zdefiniować ForEach sink z obiektu writer, wywołaj dp.foreach_sink() i przekazaj obiekt z argumentem writer :

from pyspark import pipelines as dp

class RowWriter:
    def open(self, partition_id, epoch_id):
        # Optional. Return False to skip the rows of this partition.
        return True

    def process(self, row):
        # Required. Called for each row.
        ...

    def close(self, error):
        # Optional. `error` is None if processing succeeded.
        ...

dp.foreach_sink(name="<name>", writer=RowWriter())

Możesz też przekazać funkcję do dp.foreach_sink() z argumentem func zamiast używać dekoratora, na przykład dp.foreach_sink(name="<name>", func=row_handler).

Parameters

Parametr	Description
wiersz	Required. A Row zawiera jeden zapis strumienia. Twoja praca, czyli process metoda pisarza, otrzymuje jeden wiersz dla każdego wezwania.
name	Optional. Unikatowa nazwa identyfikująca ujście w potoku. Domyślnie przyjmuje nazwę funkcji. Dla obiektu writer domyślnie używa nazwy klasy małymi literami.
func	Optional. Funkcja, którą potok wywołuje dla każdego wiersza. Używaj, func gdy dzwonisz bezpośrednio dp.foreach_sink() , a nie jako dekorator. Nie możesz przejść obu i .funcwriter
writer	Optional. Obiekt implementujący metodę process(row) , a opcjonalnie open(partition_id, epoch_id) metody i close(error) . Przejdź to jako argument na podstawie słów kluczowych. Nie możesz przejść obu i .funcwriter

Metody pisania

Obiekt writer wykorzystuje te same metody co Apache Spark Structured Streaming foreach writer. Dla każdej partycji każdej mikropartii potok wywołuje open, następnie wywołuje process dla każdego wiersza , jeśli open zwraca True, a następnie wywołuje close. Aby uzyskać informacje o pisaniu niestandardowych wyjść w Structured Streaming, zobacz Użyj foreachBatch do zapisu do dowolnych pochłanięć danych.

Metoda	Description
open(partition_id, epoch_id)	Optional. Potok wywołuje tę metodę przed przetworzeniem wierszy partycji dla mikropartii. Używaj go do otwierania połączeń lub rozpoczynania transakcji. Powrót, True aby przetworzyć wiersze partycji lub False je pominąć. Jeśli pisacz nie zdefiniuje open, pipeline przetwarza wiersze każdej partycji. partition_id i epoch_id może powtarzać się w przepływach, które zapisują się do tego samego sinka i po pełnym odświeżeniu. Zanim użyjesz ich do deduplikacji wierszy, zobacz Obsługa błędów i duplikowanie wierszy.
process(row)	Required. Potok wywołuje tę metodę dla każdego wiersza.
close(error)	Optional. Potok wywołuje tę metodę po przetworzeniu wierszy partycji. error jest wyjątkiem występującym podczas przetwarzania wierszy lub None jeśli przetwarzanie zakończyło się sukcesem. Użyj go do zamykania kontaktów. Jeśli open rozpoczęto transakcję, zatwierdzaj ją tylko wtedy, gdy error jest , Nonea w przeciwnym razie cofnij ją.