foreach_sink

El @dp.foreach_sink() decorador define un ForEach sumidero, que procesa un flujo fila a fila con lógica Python personalizada. Referencias el sumidero como un target flujo de adición o un flujo de actualización para escribir los datos. Para orientación conceptual, consideraciones y ejemplos, véase Usar un ForEach sumidero para escribir cada fila en pipelines. Para procesar cada microlote como un DataFrame, véase foreach_batch_sink.

Syntax

Para definir un ForEach sumidero a partir de una función, decora la función 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.

Para definir un ForEach sumidero a partir de un objeto escritor, llama dp.foreach_sink() y pasa el objeto con el writer argumento:

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

También puedes pasar una función a dp.foreach_sink() con el func argumento en lugar de usar el decorador, por dp.foreach_sink(name="<name>", func=row_handler)ejemplo .

Parameters

Parameter	Description
fila	Required. Un Row que contiene un registro del stream. Tu trabajo, o el process método de tu redactor, recibe una fila por cada llamada.
name	Optional. Un nombre único para identificar el destino dentro de la canalización. Por defecto se asigna al nombre de la función. Para un objeto escritor, por defecto el nombre de la clase está en minúscula.
func	Optional. La función que llama la tubería para cada fila. Úsalo func cuando llames dp.foreach_sink() directamente en lugar de como decorador. No puedes aprobar tanto func como writer.
writer	Optional. Un objeto que implementa el process(row) método, y opcionalmente los open(partition_id, epoch_id) métodos y close(error) . Pásalo como un argumento de palabras clave. No puedes aprobar tanto func como writer.

Métodos de escritura

Un objeto writer utiliza los mismos métodos que el escritor Apache Spark Structured Streaming foreach . Para cada partición de cada microlote, la tubería llama opena , luego llama process a cada fila si open devuelve True, y luego llama closea . Para información sobre cómo escribir salidas personalizadas en Structured Streaming, véase Usar foreachBatch para escribir en sumideros de datos arbitrarios.

Method	Description
open(partition_id, epoch_id)	Optional. La tubería llama a este método antes de procesar las filas de una partición para un microlote. Úsalo para abrir conexiones o iniciar transacciones. Vuelve True a procesar las filas de la partición, o False para omitirlas. Si el escritor no define open, la tubería procesa las filas de cada partición. partition_id y epoch_id puede repetirse entre flujos que escriben en el mismo sumidero y tras una actualización completa. Antes de usarlos para deduplicar filas, consulta Gestionar fallos y duplicar filas.
process(row)	Required. La tubería llama a este método para cada fila.
close(error)	Optional. La tubería llama a este método después de procesar las filas de la partición. error es la excepción que ocurre al procesar filas, o None si el procesamiento tuvo éxito. Úsalo para cerrar conexiones. Si open inicias una transacción, haz commit solo cuando error sea None, y reculézala en caso contrario.