foreach_sink

Decoratorn @dp.foreach_sink() definierar en ForEach sink, som bearbetar en ström en rad i taget med anpassad Python-logik. Du refererar till sinken som ett target i ett appendflöde eller ett uppdateringsflöde för att skriva datan. För konceptuell vägledning, överväganden och exempel, se Använd en ForEach sänka för att skriva varje rad i pipelines. För att bearbeta varje mikrobatch som en DataFrame istället, se foreach_batch_sink.

Syntax

För att definiera en ForEach sänka från en funktion, dekorera funktionen med @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.

För att definiera en ForEach sänka från ett writer-objekt, anropa dp.foreach_sink() och skicka objektet med argumentet 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())

Du kan också skicka en funktion till dp.foreach_sink() med argumentet func istället för att använda dekoratören, till exempel dp.foreach_sink(name="<name>", func=row_handler).

Parameters

Parameter	Description
rad	Required. A Row som innehåller en post av strömmen. Din funktion, eller process metoden för din författare, får en rad för varje anrop.
name	Optional. Ett unikt namn för att identifiera slutpunkten i pipelinen. Fungerar som standard på funktionens namn. För ett writer-objekt går klassens namn med gemener som standard.
func	Optional. Funktionen som pipelinen anropar för varje rad. Använd func när du ringer dp.foreach_sink() direkt istället för som dekoratör. Du kan inte klara både func och writer.
writer	Optional. Ett objekt som implementerar metoden process(row) , och valfritt och open(partition_id, epoch_id)close(error) metoderna. Ta det som ett nyckelordsargument. Du kan inte klara både func och writer.

Skrivarmetoder

Ett writer-objekt använder samma metoder som Apache Spark Structured Streaming-skrivaren foreach . För varje partition av varje mikrobatch anropar openpipelinen , sedan anropar process för varje rad om open returnerar True, och sedan anropar close. För information om att skriva anpassade utdata i Structured Streaming, se Använd foreachBatch för att skriva till godtyckliga datasänkor.

Method	Description
open(partition_id, epoch_id)	Optional. Pipelinen anropar denna metod innan den bearbetar raderna i en partition för en mikrobatch. Använd den för att öppna anslutningar eller starta transaktioner. Gå tillbaka True för att bearbeta raderna i partitionen, eller False för att hoppa över dem. Om författaren inte definierar open, bearbetar pipelinen raderna i varje partition. partition_id och epoch_id kan upprepas över flöden som skriver till samma sänka och efter en fullständig uppdatering. Innan du använder dem för att deduplicera rader, se Hantera misslyckanden och duplicera rader.
process(row)	Required. Pipelinen anropar denna metod för varje rad.
close(error)	Optional. Pipelinen anropar denna metod efter att den bearbetat raderna i partitionen. error är undantaget som inträffade under bearbetning av rader, eller None om bearbetningen lyckades. Använd den för att stänga anslutningar. Om en open transaktion startas, commit den bara när error är None, och rulla tillbaka den annars.