One Pipeline, Any Destination: The ForEachBatch Sink in Spark Declarative Pipelines is now GA

[john_armstrong]

TL;DR

Until recently, a Spark Declarative Pipeline’s (SDP) jurisdiction ended where the lakehouse ended. You could land, clean, and curate data into governed tables in Unity Catalog, and that was the contract. Use cases that required streaming to a system without a first-party connector (a SaaS app that only speaks REST, a search index that takes document-shaped JSON, etc.) fell outside the scope of SDP. In practice, this gap was filled by a standalone Structured Streaming job bolted onto the declarative one, resulting in a second framework and an additional component to maintain.

The standalone job exists because Structured Streaming's foreachBatch can do almost anything. Inside a streaming query, you get a DataFrame and a batch ID, and you can call a REST endpoint, run a MERGE with custom CDC rules, or apply a Python enrichment step that's too imperative for SQL. The catch is everything outside the write: a production pipeline also needs checkpoints, retries, idempotency, data quality, and monitoring, and a standalone job makes all of that yours to build and operate.

The ForEachBatch sink in Spark Declarative Pipelines closes that gap and is Generally Available as of May 2026.

• Write streaming data to any destination from inside a managed pipeline: REST endpoints, search and vector indexes, CRMs, multiple Delta tables, and multi-topic Kafka routing.
• Decrease operational cost by maintaining and monitoring a single pipeline and framework, rather than a blend of SDP and Structured Streaming jobs.
• Advanced building blocks when you need them: per-flow idempotent writes, custom CDC, and cross-table atomicity with Multi-Statement Transactions (Public Preview).

How ForEachBatch Sinks Work

An SDP ForEachBatch sink is a Python function that the pipeline calls once per micro-batch of input rows. The decorator @sdp.foreach_batch_sink(name="...") registers the function as a named sink in the pipeline graph. The framework owns the streaming loop, the checkpoints, and the retry behavior, and the function body owns the transform and write. The runtime does not inspect or constrain what the body does. The parameters are (df, batch_id), where df holds the rows in the current micro-batch and batch_id is a monotonically increasing integer that the engine assigns to each batch and resets to zero on a full refresh.

Inside the body, you have the full Python runtime, so the transformation and writing can be anything you can do with Python. The streaming source is defined by one or more @sdp.append_flow(target="...") definitions that point to the ForEachBatch sink, turning each flow into an independent streaming query. Each query has its own checkpoint, so a slow flow does not block a fast one, and a failure in one flow does not impact the checkpoint of another. SDP creates and manages the checkpoint paths. There is no checkpointLocation option to set.

The example below illustrates a job that most data teams take on sooner or later: cleaning messy, user-entered contact data before it reaches a curated table. Parsing phone numbers is notoriously hard to get right in SQL and a solved problem in Python, so the transform wraps the phonenumbers library (installed through the pipeline environment) in a UDF. The UDF is not what calls for ForEachBatch here. Streaming pipelines can already apply a UDF to a DataFrame column and write the result to a table. What ForEachBatch grants is the ability to write to multiple sinks from a single microbatch. Rows that normalize cleanly belong in the curated table, and the rest in a quarantine table for review.  A declarative pipeline expresses this as two tables, each reading the source and re-running the parse. The sink does the same work in one pass, parsing once and writing each side as an idempotent append.

import phonenumbers

from pyspark import pipelines as sdp
from pyspark.sql import functions as F

@F.udf("string")
def to_e164(raw):
    if not raw:
        return None
    try:
        parsed = phonenumbers.parse(raw, "US")
        return phonenumbers.format_number(
            parsed, phonenumbers.PhoneNumberFormat.E164)
    except phonenumbers.NumberParseException:
        return None

@sdp.foreach_batch_sink(name="contact_cleanup_sink")
def clean_contacts(df, batch_id):
    cleaned = df.withColumn("phone_e164", to_e164("phone_raw")).persist()
    (cleaned.where("phone_e164 IS NOT NULL").write
        .option("txnVersion", batch_id).option("txnAppId", "contact_cleanup_clean")
        .mode("append").saveAsTable("main.crm.contacts_clean"))
    (cleaned.where("phone_e164 IS NULL").write
        .option("txnVersion", batch_id).option("txnAppId", "contact_cleanup_quarantine")
        .mode("append").saveAsTable("main.crm.contacts_quarantine"))
    cleaned.unpersist()

@sdp.append_flow(target="contact_cleanup_sink")
def raw_contacts_flow():
    return spark.readStream.table("main.crm.raw_contacts")

In this example, both appends are idempotent because they use the ForEachBatch input batch_id as the txnVersion, allowing the stream to maintain a state of batches that have been successfully committed for each txnAppId.  Without these configurations, a failure during processing that results in a batch being committed to one sink rather than the other would lead to duplicate data when the batch is retried.

A few more contract details are worth knowing up front. First, if the function writes to more than one destination, persist or cache the DataFrame first, or each Spark action against df re-reads the upstream source. Second, the function body runs on the driver as coordination code, while the DataFrame operations inside it run as ordinary distributed Spark jobs. The function must remain serializable so that SDP can ship it to the pipeline's workers. In practice, that just means you shouldn't close over live objects like open connections or sockets; instead, build those inside the body and close over plain config and secrets (a non-serializable function draws a WARN in the event log, not a hard failure). That's also why dbutils is not available inside it, so read secrets and parameters at definition time and close over them. Third, ForEachBatch is also a streaming-only sink, so flows must be streaming sources, and AUTO CDC semantics are not supported inside the body.

What SDP handles for you: per-flow checkpoint creation and recovery, source-offset tracking, replay on retry, UC governance and lineage, an event-log entry per batch, and pipeline-level retries on transient failures.

What you handle: idempotency for non-Delta destinations, downstream cleanup on full refresh, connection management for external systems (pooling, auth, backoff), and schema enforcement on destinations that don't have it.

When using ForEachBatch, you need to consider how a full pipeline refresh affects the sink's output.  During full refresh, SDP resets its checkpoints so that every input row is replayed through the sink function. If the downstream system is not idempotent, you own the reset step before the rerun. There’s one trap to avoid here: for Delta targets written with the txnVersion/txnAppId pattern, that reset means dropping and recreating the table (or rotating the txnAppId). Delta's transaction state survives TRUNCATE, so a truncated table will silently skip every replayed batch as an already-seen duplicate and stay empty.

How OpenAI's Security Team Turned a Streaming Pipeline Into Its Own Latency Watchdog with SDP ForEachBatch

OpenAI's security team runs an SDP pipeline that processes telemetry from the Kubernetes clusters hosting training and inference workloads. The data lands in governed Delta tables for downstream investigation. A Delta table, though, is a data contract, not an operational contract. The team also needed to know, in near real time, how fresh the telemetry was on a per-cluster basis, and to alert when a given cluster's data started lagging. A lagging cluster is a blind spot for security.

The solution is a ForEachBatch sink inside the same SDP pipeline as the main telemetry flow. Each micro-batch is grouped by table, tier, and cluster with ingestion and event lag percentiles computed in a single groupBy().agg() then POSTed to the observability platform over REST. Failures inside the sink are caught and logged, so a hiccup in the observability stack can never break primary ingestion.

We built our observability platform on SDP because we needed a data engineering framework that let us declare the datasets we need with minimal overhead for tuning performance, data quality, and infrastructure. With ForEachBatch support, even the most arbitrary pipelines run inside SDP, too, which means a simpler architecture, improved operations, and better observability.

Jiadong Zhang, Member of Technical Staff @ OpenAI

import requests

from pyspark import pipelines as sdp
from pyspark.sql import functions as F

METRICS_ENDPOINT = "https://observability.example.com/api/v1/metrics"

# Read once at definition time. dbutils is not available inside the sink body.

API_TOKEN = dbutils.secrets.get(scope="observability", key="api_token")

def post_latency_metrics(rows, batch_id):
    payload = [{**row.asDict(), "batch_id": batch_id} for row in rows]
    requests.post(
        METRICS_ENDPOINT, json=payload, timeout=10,
        headers={"Authorization": f"Bearer {API_TOKEN}"},
    ).raise_for_status()

@sdp.foreach_batch_sink(name="security_events_latency_metrics_sink")
def emit_latency_metrics(batch_df, batch_id):
    try:
        if batch_df.isEmpty():
            return
        now_s = F.unix_timestamp(F.current_timestamp())
        pcts = F.array(F.lit(0.01), F.lit(0.5), F.lit(0.9), F.lit(0.99))
        latency_df = (
            batch_df
            .withColumn("ingestion_lag_s", now_s - F.unix_timestamp(F.col("etl_ingestion_timestamp")))
            .withColumn("event_lag_s", now_s - F.unix_timestamp(F.col("_event_timestamp")))
        )
        rows = (
            latency_df.where(F.col("cluster_name").isNotNull())
            .groupBy("cluster_name")
            .agg(
                F.percentile_approx("ingestion_lag_s", pcts).alias("ingestion_lag_p"),
                F.percentile_approx("event_lag_s", pcts).alias("event_lag_p"),
            )
            .collect()
        )
        post_latency_metrics(rows, batch_id)
    except Exception as e:
        # Observability must never break primary ingestion.
        print(f"[latency-metrics] batch {batch_id} failed: {e}")

@sdp.append_flow(target="security_events_latency_metrics_sink")
def latency_metrics_flow():
    return (spark.readStream.table("security_events_final")
            .select("cluster_name", "etl_ingestion_timestamp", "_event_timestamp"))

The alert that fires when a cluster's latency drifts past its SLO is now driven by the same pipeline that produces the underlying data, and is implemented and managed by a single data engineering framework.

Beyond the Basics

The worked example covers the headline pattern: a destination with no first-party connector, written from inside one governed pipeline. Three more capabilities deserve a mention:

• Custom CDC and cross-table atomicity. AUTO CDC covers most CDC scenarios. For the rest (multi-source MERGE, soft deletes, business-key dedup), write the DeltaTable.merge() yourself inside the sink body. When two tables must commit together, Multi-Statement Transactions work within the sink (Public Preview on UC-managed Delta; Private Preview on UC-managed Iceberg).
• One-pass fan-out. Persist the batch DataFrame once and write it to N destinations, instead of running N parallel streaming jobs that each re-read the upstream source.
• Exactly-once to external systems. Every batch arrives with batch_id for exactly this purpose: pass it as txnVersion/txnAppId for Delta sinks, or map it onto whatever the destination's protocol offers (a Kafka transactional id, a REST idempotency header, a JDBC unique-key upsert).

Getting Started

ForEachBatch sinks are generally available today. If you're already running on Databricks, head to the ForEachBatch sink documentation for the full API reference and runnable examples, and add a sink to an existing pipeline with a single decorated function. New to Databricks? Start a free trial and build your first declarative pipeline in minutes. If you want to go deeper, the runtime release notes cover per-flow checkpointing and the idempotent-write helpers, the Iceberg interop docs cover managed Iceberg destinations, and the Multi-Statement Transactions docs cover enrollment for cross-table atomicity.  You can also find the examples shared in this blog in our databricks-blogposts GitHub repository here.