foreach_sink

Der Dekorateur @dp.foreach_sink() definiert eine ForEach Senke, die einen Strom Reihe für Reihe mit benutzerdefinierter Python-Logik verarbeitet. Du bezugst den Sink in target einem Anhängungsfluss oder als Update-Fluss, um die Daten zu schreiben. Für konzeptionelle Anleitungen, Überlegungen und Beispiele siehe Verwenden Sie eine ForEach Senke, um jede Zeile in Pipelines zu schreiben. Um jede Mikrocharge stattdessen als DataFrame zu verarbeiten, siehe foreach_batch_sink.

Syntax

Um eine ForEach Senke aus einer Funktion zu definieren, dekoriert man die Funktion mit @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.

Um einen ForEach Sink aus einem Writer-Objekt zu definieren, ruft dp.foreach_sink() und übergibt man das Objekt mit dem Argument 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 kannst auch eine Funktion mit dem func Argument anstelle des dp.foreach_sink() Decorators weitergeben, zum Beispiel dp.foreach_sink(name="<name>", func=row_handler).

Parameters

Parameter	Description
rudern	Required. A Row , der einen Datensatz des Streams enthält. Ihre Funktion, oder die process Methode Ihres Autors, erhält für jeden Aufruf eine Zeile.
Name	Optional. Ein eindeutiger Name zum Identifizieren der Spüle innerhalb der Pipeline. Standardmäßig wird der Name der Funktion verwendet. Für ein Writer-Objekt wird standardmäßig der Klassenname in Kleinbuchstaben verwendet.
func	Optional. Die Funktion, die die Pipeline für jede Zeile aufruft. Nutze func es, wenn du direkt anrufst dp.foreach_sink() , statt als Dekorateur. Man kann nicht sowohl func bestehen als auch writer.
writer	Optional. Ein Objekt, das die Methode process(row) implementiert und optional die open(partition_id, epoch_id) und-Methoden close(error) . Nehmen Sie es als Keyword-Argument aus. Man kann nicht sowohl func bestehen als auch writer.

Schreibmethoden

Ein Writer-Objekt verwendet die gleichen Methoden wie der Apache Spark Structured Streaming Writer foreach . Für jede Partition jedes Mikro-Batches ruft opendie Pipeline , dann ruft process sie für jede Zeile , wenn open , Trueund dann ruft close. Informationen zum Schreiben benutzerdefinierter Ausgaben in Structured Streaming finden Sie unter Use foreachBatch, um auf beliebige Datensenken zu schreiben.

Method	Description
open(partition_id, epoch_id)	Optional. Die Pipeline ruft diese Methode auf, bevor sie die Zeilen einer Partition für einen Mikro-Batch verarbeitet. Nutze es, um Verbindungen zu öffnen oder Transaktionen zu starten. Zurück, True um die Zeilen der Partition zu verarbeiten oder False sie zu überspringen. Wenn der Autor nicht definiert open, verarbeitet die Pipeline die Zeilen jeder Partition. partition_id und epoch_id kann sich über Flows hinweg wiederholen, die in denselben Sink schreiben, und nach einer vollständigen Aktualisierung. Bevor du sie zum Deduplizieren von Zeilen verwendest, siehe Handle failures und duplicate rows.
process(row)	Required. Die Pipeline ruft diese Methode für jede Zeile auf.
close(error)	Optional. Die Pipeline ruft diese Methode auf, nachdem sie die Zeilen der Partition verarbeitet hat. error ist die Ausnahme, die während der Verarbeitung von Zeilen aufgetreten ist oder None wenn die Verarbeitung erfolgreich war. Nutze es, um Verbindungen zu schließen. Wenn open eine Transaktion gestartet ist, committiere sie nur, wenn error es ist None, und rücke sie ansonsten zurück.