foreach_sink

Il @dp.foreach_sink() decorator definisce un ForEach sink, che elabora un flusso riga alla volta con una logica Python personalizzata. Fai riferimento al sink come un target flow add o un flow di aggiornamento per scrivere i dati. Per indicazioni concettuali, considerazioni ed esempi, vedi Usa un ForEach sink per scrivere ogni riga nelle pipeline. Per elaborare ogni micro-batch come DataFrame, vedi foreach_batch_sink.

Syntax

Per definire un ForEach pozzo da una funzione, decorare la funzione con @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.

Per definire un ForEach sink da un oggetto writer, chiama dp.foreach_sink() e passa l'oggetto con l'argomento 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())

Puoi anche passare una funzione a dp.foreach_sink() con l'argomento func invece di usare il decoratore, ad esempio dp.foreach_sink(name="<name>", func=row_handler).

Parameters

Parametro	Description
row	Required. A Row che contiene un record del flusso. Il tuo lavoro, o il process metodo del tuo scrittore, riceve una riga per ogni chiamata.
name	Optional. Nome univoco per identificare il sink all'interno della pipeline. Di default si attribuisce al nome della funzione. Per un oggetto scrittore, il nome della classe viene scritto in minuscolo.
func	Optional. La funzione che la pipeline chiama per ogni riga. Usalo func quando chiami dp.foreach_sink() direttamente invece che come decoratore. Non puoi superare entrambi func e writer.
writer	Optional. Un oggetto che implementa il process(row) metodo e, opzionalmente, i open(partition_id, epoch_id) metodi e close(error) . Passalo come argomento per parola chiave. Non puoi superare entrambi func e writer.

Metodi di scrittura

Un oggetto writer utilizza gli stessi metodi dello writer Apache Spark Structured Streaming foreach . Per ogni partizione di ogni micro-batch, la pipeline chiama open, poi chiama process per ogni riga se open restituisce True, e poi chiama close. Per informazioni sulla scrittura di output personalizzati nello Structured Streaming, vedi Usa foreachBatch per scrivere in data sink arbitrari.

Metodo	Description
open(partition_id, epoch_id)	Optional. La pipeline chiama questo metodo prima di elaborare le righe di una partizione per un micro-batch. Usalo per aprire connessioni o avviare transazioni. Torna True a elaborare le righe della partizione, oppure False a saltarle. Se lo scrittore non definisce open, la pipeline processa le righe di ogni partizione. partition_id e epoch_id può ripetersi tra flussi che scrivono nello stesso sink e dopo un aggiornamento completo. Prima di usarli per deduplicare righe, consulta Gestire fallimenti e duplicate righe.
process(row)	Required. La pipeline chiama questo metodo per ogni riga.
close(error)	Optional. La pipeline chiama questo metodo dopo aver elaborato le righe della partizione. error è l'eccezione che si verifica durante l'elaborazione delle righe, o None se l'elaborazione è riuscita. Usalo per chiudere le connessioni. Se open inizia una transazione, commmitla solo quando error è None, e altrimenti annulla indietro.