foreach_sink

The @dp.foreach_sink() decorator defines a ForEach sink, which processes a stream one row at a time with custom Python logic. You reference the sink as a target in an append flow or an update flow to write the data. For conceptual guidance, considerations, and examples, see Use a ForEach sink to write each row in pipelines. To process each micro-batch as a DataFrame instead, see foreach_batch_sink.

Syntax

To define a ForEach sink from a function, decorate the function with @dp.foreach_sink():

Python

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.

To define a ForEach sink from a writer object, call dp.foreach_sink() and pass the object with the writer argument:

Python

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

You can also pass a function to dp.foreach_sink() with the func argument instead of using the decorator, for example dp.foreach_sink(name="<name>", func=row_handler).

Parameters

Parameter	Description
row	Required. A Row that contains one record of the stream. Your function, or the process method of your writer, receives one row for each call.
name	Optional. A unique name to identify the sink within the pipeline. Defaults to the name of the function. For a writer object, defaults to the class name in lowercase.
func	Optional. The function that the pipeline calls for each row. Use func when you call dp.foreach_sink() directly instead of as a decorator. You can't pass both func and writer.
writer	Optional. An object that implements the process(row) method, and optionally the open(partition_id, epoch_id) and close(error) methods. Pass it as a keyword argument. You can't pass both func and writer.

Parameter	Description
row	Required. A Row that contains one record of the stream. Your function, or the process method of your writer, receives one row for each call.
name	Optional. A unique name to identify the sink within the pipeline. Defaults to the name of the function. For a writer object, defaults to the class name in lowercase.
func	Optional. The function that the pipeline calls for each row. Use func when you call dp.foreach_sink() directly instead of as a decorator. You can't pass both func and writer.
writer	Optional. An object that implements the process(row) method, and optionally the open(partition_id, epoch_id) and close(error) methods. Pass it as a keyword argument. You can't pass both func and writer.

Writer methods

A writer object uses the same methods as the Apache Spark Structured Streaming foreach writer. For each partition of each micro-batch, the pipeline calls open, then calls process for each row if open returns True, and then calls close. For information about writing custom outputs in Structured Streaming, see Use foreachBatch to write to arbitrary data sinks.

Method	Description
open(partition_id, epoch_id)	Optional. The pipeline calls this method before it processes the rows of a partition for a micro-batch. Use it to open connections or start transactions. Return True to process the rows of the partition, or False to skip them. If the writer doesn't define open, the pipeline processes the rows of each partition. partition_id and epoch_id can repeat across flows that write to the same sink and after a full refresh. Before you use them to deduplicate rows, see Handle failures and duplicate rows.
process(row)	Required. The pipeline calls this method for each row.
close(error)	Optional. The pipeline calls this method after it processes the rows of the partition. error is the exception that occurred while processing rows, or None if processing succeeded. Use it to close connections. If open started a transaction, commit it only when error is None, and roll it back otherwise.

Method	Description
open(partition_id, epoch_id)	Optional. The pipeline calls this method before it processes the rows of a partition for a micro-batch. Use it to open connections or start transactions. Return True to process the rows of the partition, or False to skip them. If the writer doesn't define open, the pipeline processes the rows of each partition. partition_id and epoch_id can repeat across flows that write to the same sink and after a full refresh. Before you use them to deduplicate rows, see Handle failures and duplicate rows.
process(row)	Required. The pipeline calls this method for each row.
close(error)	Optional. The pipeline calls this method after it processes the rows of the partition. error is the exception that occurred while processing rows, or None if processing succeeded. Use it to close connections. If open started a transaction, commit it only when error is None, and roll it back otherwise.