Usa un ForEach sink per scrivere ogni riga nelle pipeline

Usa un ForEach sink per scrivere dati in streaming verso destinazioni che non hanno un sink integrato, come un'API REST esterna o uno store chiave-valore.

Il ForEach sink processa un flusso una riga alla volta. Per ogni riga, la pipeline richiama una funzione Python personalizzata o un oggetto writer, simile al foreach sink in Apache Spark Structured Streaming.

Note

Il ForEach sink descritto in questa pagina è per le pipeline. Apache Spark Structured Streaming supporta anche foreach. Per informazioni sulla scrittura di output personalizzati nello Structured Streaming, vedi Usa foreachBatch per scrivere in data sink arbitrari.

Quando usare un ForEach sink

Usa un ForEach sink quando devi scrivere ogni record singolarmente in una destinazione che non ha un formato sink integrato, ad esempio chiamando un'API esterna o aggiornando uno store chiave-valore per ogni record. Usa invece un ForEachBatch sink quando la tua logica funziona su un intero micro-batch, ad esempio eseguendo il merge in una tabella Delta Lake o scrivendo in più destinazioni con operazioni DataFrame.

Questa tabella confronta i due sink:

Caratteristica ForEach Lavandino ForEachBatch Lavandino
Unità di elaborazione Una fila alla volta Un DataFrame micro-batch alla volta
Dove gira il tuo codice Sugli esecutori Nel driver
API Python @dp.foreach_sink() @dp.foreach_batch_sink()

Per informazioni sui sink integrati, vedi Sinks in Lakeflow pipelines. Per il @dp.foreach_sink() riferimento Python API, vedi foreach_sink.

Definisci un ForEach sink

Puoi definire un ForEach sink da una funzione o da un oggetto writer.

Per definire il sink da una funzione, decorare la funzione con @dp.foreach_sink(). La pipeline chiama la funzione per ogni riga. Se ometti name, il sink usa il nome della funzione:

from pyspark import pipelines as dp

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

Per definire il sink a partire da un oggetto writer, passa l'oggetto con l'argomento parola chiave writer. Il writer deve implementare process(row) e può implementare open(partition_id, epoch_id) e close(error). Usa un writer quando vuoi aprire una connessione per ogni partizione e riutilizzarlo per ogni riga. Se ometti name, il sink usa il nome della classe dello scrittore, in minuscolo:

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())

Per dettagli su ciascun metodo, vedi metodi di Writer.

Voci del registro eventi

Quando definisci un ForEach sink, il log degli eventi della pipeline registra un SinkDefinition evento con "format": "foreach" per il sink.

Scrivi in un ForEach sink

Puoi scrivere su un ForEach sink da un append flow o da un update flow.

Scrivi in modalità append

Per scrivere in un ForEach sink da un flusso append, fare riferimento al nome del sink come segue: target

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

Puoi definire più flussi di append che puntano allo stesso ForEach sink. Ogni flusso mantiene il proprio checkpoint.

Scrivi usando un flusso di aggiornamento

Per scrivere in un ForEach sink da un flusso di aggiornamento, usa il @dp.update_flow decorator:

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

Un flusso di aggiornamento può inviare la stessa chiave al sink di nuovo in un micro-batch successivo con un nuovo valore. Per sostituire il vecchio valore, esegui l'upsert di ogni riga nella destinazione per chiave invece di inserirla. Nell'esempio precedente, usa user_id come chiave, in modo che ogni nuovo conteggio per un utente sostituisca il conteggio precedente.

Per l'elaborazione a bassa latenza, un flusso di aggiornamento che scrive in un sink ForEach può essere eseguito in modalità in tempo reale. Vedere Usare la modalità in tempo reale nelle pipeline di Lakeflow.

Importazione di librerie e dipendenze sugli esecutori

La tua funzione o writer funziona sugli esecutori, non sul driver. Seguire queste procedure:

  • Aggiungi i pacchetti di cui il tuo codice ha bisogno all'ambiente della pipeline. I pacchetti nell'ambiente pipeline sono disponibili per il tuo codice sugli esecutori. Consulta Gestione delle dipendenze Python per le pipeline.
  • Importa le librerie che il tuo codice usa all'interno della funzione o dei metodi di scrittura, non solo in cima al file sorgente della pipeline.
  • Non affidarti agli oggetti che crei sul driver, come un client o una connessione che apri in cima al file. Scrivili all'interno della funzione o nel metodo open dello scrittore.

Gestire gli errori e le righe duplicate

Se il tuo codice genera un'eccezione, il flusso fallisce e il log eventi della pipeline registra un evento di errore che include il messaggio di eccezione.

Quando un micro-batch viene riprovato dopo un errore, il sink può ricevere nuovamente alcune righe. Rendi idempotenti le tue operazioni di scrittura per riga, ad esempio usando una chiave univoca della riga per eseguire un upsert nella destinazione. Consulta Garanzie di elaborazione nelle pipeline Lakeflow.

Se usi un oggetto writer, open riceve un partition_id e un epoch_id. Insieme, questi ID identificano le righe di una partizione in un micro-batch, così puoi usarli per deduplicare le righe. Una funzione non riceve questi ID.

Gli ID sono unici solo all'interno del checkpoint di un determinato flusso. La stessa coppia può ripetersi nei seguenti casi:

  • Diversi flussi scrivono nello stesso sink.
  • Un ricaricamento completo resetta i checkpoint.

Per deduplicare con questi ID, scrivi nel sink da un singolo flusso e cancella lo stato di deduplicazione quando esegui un refresh completo.

Aggiornamento completo

Con un aggiornamento completo, la pipeline resetta il checkpoint di ogni flusso che scrive nel ForEach sink. La pipeline non traccia né ripulisce i dati che il sink scrive, quindi gestisci la retention e la pulizia per la destinazione. Se hai bisogno di un nuovo inizio, rimuovi quei dati da solo.

Limitations

  • Metriche di riga: La pipeline non riporta il numero di righe che un ForEach sink scrive.
  • Chiamate per riga: Se il tuo codice chiama un sistema esterno per ogni riga, il sistema esterno può limitare il throughput. Per ridurre il sovraccarico, apri le connessioni nel metodo open dello scrittore e riutilizzale per ogni riga, oppure usa un sink ForEachBatch per scrivere in massa.

Examples

Questi esempi mostrano modi comuni di usare un sink ForEach.

Esempio della sintassi di base

Questo esempio definisce un ForEach sink da una funzione e vi scrive da un flusso 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")

Scrivi ogni riga verso un'API esterna

Questo esempio utilizza un oggetto writer per inviare ogni ordine a un'API REST esterna. Il writer apre una sessione HTTP per ogni partizione. Utilizza l'ID dell'ordine nell'URL, così una riga riprovata aggiorna lo stesso record invece di creare un duplicato:

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")

Scrivere in un solo sink da più flussi

Questo esempio scrive due fonti nello stesso ForEach sink:

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")

Risorse aggiuntive

  • Ultimo aggiornamento: