Use a ForEach sink to write each row in pipelines
Use a ForEach sink to write streaming data to destinations that don't have a built-in sink, such as an external REST API or a key-value store.
The ForEach sink processes a stream one row at a time. For each row, the pipeline calls a custom Python function or a writer object, similar to the foreach sink in Apache Spark Structured Streaming.
The ForEach sink described on this page is for pipelines. Apache Spark Structured Streaming also supports foreach. For information about writing custom outputs in Structured Streaming, see Use foreachBatch to write to arbitrary data sinks.
When to use a ForEach sink
Use a ForEach sink when you need to write each record individually to a destination that has no built-in sink format, such as calling an external API or updating a key-value store for each record. Use a ForEachBatch sink instead when your logic works on a whole micro-batch, such as merging into a Delta Lake table or writing to several destinations with DataFrame operations.
This table compares the two sinks:
Characteristic |
|
|
|---|---|---|
Unit of processing | One row at a time | One micro-batch DataFrame at a time |
Where your code runs | On the executors | On the driver |
Python API |
|
|
For information about built-in sinks, see Sinks in Lakeflow pipelines. For the @dp.foreach_sink() Python API reference, see foreach_sink.
Define a ForEach sink
You can define a ForEach sink from a function or from a writer object.
To define the sink from a function, decorate the function with @dp.foreach_sink(). The pipeline calls the function for each row. If you omit name, the sink uses the function name:
from pyspark import pipelines as dp
@dp.foreach_sink()
def my_row_sink(row):
# Custom per-row logic here.
...
To define the sink from a writer object, pass the object with the writer keyword argument. The writer must implement process(row) and can implement open(partition_id, epoch_id) and close(error). Use a writer when you want to open one connection for each partition and reuse it for each row. If you omit name, the sink uses the class name of the writer, in lowercase:
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())
For details about each method, see Writer methods.
Event log entries
When you define a ForEach sink, the pipeline event log records a SinkDefinition event with "format": "foreach" for the sink.
Write to a ForEach sink
You can write to a ForEach sink from an append flow or an update flow.
Write with an append flow
To write to a ForEach sink from an append flow, reference the sink name as the target:
@dp.append_flow(target="my_row_sink")
def my_flow():
return spark.readStream.table("my_catalog.my_schema.source_table")
You can define multiple append flows that target the same ForEach sink. Each flow keeps its own checkpoint.
Write with an update flow
To write to a ForEach sink from an update flow, use the @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()
An update flow can send the same key to the sink again in a later micro-batch with a new value. To replace the old value, upsert each row into your destination by key instead of inserting it. In the preceding example, use user_id as the key, so that each new count for a user replaces the previous count.
For low-latency processing, an update flow that writes to a ForEach sink can run in real-time mode. See Use real-time mode in Lakeflow pipelines.
Import libraries and dependencies on the executors
Your function or writer runs on the executors, not on the driver. Follow these practices:
- Add the packages that your code needs to the pipeline environment. Packages in the pipeline environment are available to your code on the executors. See Manage Python dependencies for pipelines.
- Import the libraries that your code uses inside the function or the writer methods, not only at the top of the pipeline source file.
- Don't rely on objects that you create on the driver, such as a client or connection that you open at the top of the file. Create them inside the function, or in the
openmethod of the writer.
Handle failures and duplicate rows
If your code raises an exception, the flow fails, and the pipeline event log records an error event that includes the exception message.
When a micro-batch is retried after a failure, the sink can receive some rows again. Make your per-row writes idempotent, for example by using a unique key from the row to upsert into the destination. See Processing guarantees in Lakeflow pipelines.
If you use a writer object, open receives a partition_id and an epoch_id. Together, these IDs identify the rows of one partition in one micro-batch, so you can use them to deduplicate rows. A function doesn't receive these IDs.
The IDs are unique only within one flow's checkpoint. The same pair can repeat in the following cases:
- Several flows write to the same sink.
- A full refresh resets the checkpoints.
To deduplicate with these IDs, write to the sink from a single flow, and clear your deduplication state when you run a full refresh.
Full refresh
On a full refresh, the pipeline resets the checkpoint of each flow that writes to the ForEach sink. The pipeline doesn't track or clean up the data that the sink writes, so you manage retention and cleanup for your destination. If you need a clean start, remove that data yourself.
Limitations
- Row metrics: The pipeline doesn't report the number of rows that a
ForEachsink writes. - Per-row calls: If your code calls an external system for each row, the external system can limit throughput. To reduce overhead, open connections in the
openmethod of the writer and reuse them for each row, or use aForEachBatchsink to write in bulk.
Examples
These examples show common ways to use a ForEach sink.
Basic syntax example
This example defines a ForEach sink from a function and writes to it from an append flow:
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")
Write each row to an external API
This example uses a writer object to send each order to an external REST API. The writer opens one HTTP session for each partition. It uses the order ID in the URL, so a retried row updates the same record instead of creating a duplicate:
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")
Write to one sink from multiple flows
This example writes two sources to the same 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")