Processing guarantees in Lakeflow pipelines

Retries and reruns are inevitable in any real pipeline, so this page explains the processing guarantees Lakeflow pipelines give you and how to keep the parts you write safe to re-run.

Overview

Two related properties determine whether re-running a pipeline is safe:

  • Idempotency means a pipeline produces the same result no matter how many times you run it over the same input. Re-running after a failure, backfilling a date range twice, or manually retriggering a job never creates duplicate rows or corrupts state.
  • Processing guarantee describes how many times each record affects the result. At-least-once processing guarantees every record is processed, but a failure and retry might process some records more than once, which risks duplicates. Exactly-once processing guarantees every record affects the result as if it were processed precisely one time, even across retries, with no duplicates and no gaps.

Lakeflow pipelines are idempotent by default for the pieces they manage, and give you exactly-once processing within their own managed tables. The important thing to understand is where those guarantees stop being automatic, so you can add the right safeguards at the edges of your pipeline.

How it works

Lakeflow pipelines provide exactly-once processing and idempotency for the flows they manage, and give you tools to keep the logic you write idempotent too.

Exactly-once processing for managed tables

Within managed tables, you get exactly-once processing by default. Streaming tables use Structured Streaming checkpoints combined with Delta Lake's transactional writes: each micro-batch commits its source offsets and its output together, so a retried batch after a failure either fully succeeds or is fully rolled back and retried, never partially applied twice. This holds for Auto Loader file ingestion, Kafka, Kinesis, and Azure Event Hubs reads, and AUTO CDC upserts, without any code from you.

If an at-least-once source sends the same record multiple times, the pipeline processes them as unique records, and writes all of them to your table. Removing those duplicates is your responsibility. See Deduplicate at-least-once sources.

Idempotency for reads follows from those same checkpoints. Auto Loader and streaming table checkpoints guarantee each source file or offset is processed once for state-tracking purposes, so reprocessing a pipeline update after a failure resumes from checkpoint rather than reprocessing or skipping data. You get this by using streaming tables over spark.readStream instead of hand-rolled batch loops. See Streaming tables.

Use AUTO CDC instead of hand-written MERGE

AUTO CDC INTO is inherently idempotent with respect to its keys and sequence_by. Applying the same change record twice, or applying records out of order, produces the same final state, because the pipeline uses the sequence column to decide whether an incoming row is actually newer than what's stored:

CREATE FLOW customers_cdc_flow AS AUTO CDC INTO customers_silver
FROM stream(customers_cdc_bronze)
KEYS (customer_id)
SEQUENCE BY sequence_num
STORED AS SCD TYPE 1;

If you write your own upsert logic outside of AUTO CDC (rare, but sometimes necessary for complex merge conditions), key it on a stable business key and make it safe to apply twice, for example a MERGE ... WHEN MATCHED keyed on order_id rather than a blind INSERT. For more information, see The AUTO CDC APIs: Simplify change data capture with pipelines.

Keep your own transformations idempotent

To keep logic idempotent when re-running write operations, follow these two guidelines:

  • Avoid non-deterministic transformations in materialized views. Because a materialized view can fully or incrementally recompute, avoid functions whose output depends on when they run rather than what the input is. For example, don't use current_timestamp() to compute a business value that should stay fixed once written; take the timestamp from the source event or pass it in as a parameter so recomputation produces identical output.
  • Design full refreshes to be safe. A full refresh drops and recomputes a table from scratch, which is only safe if every upstream source can still produce the full history. If an upstream source only exposes a rolling window of changes, a full refresh of a downstream AUTO CDC table can silently lose history, so design source and topic retention with this in mind.

Get exactly-once at the edges

Where exactly-once stops being automatic is at the edges of what the pipeline directly controls, such as writes to external systems. When you fan out to an external system, make the write itself idempotent, for example by upserting by key on the receiving side, since a retried micro-batch could otherwise write the same batch twice. The following sink writes each partition of the batch from the executors and uses an idempotency key so a retried batch doesn't double-write:

from pyspark import pipelines as dp

@dp.foreach_batch_sink(name="orders_to_external_api")
def write_orders_to_api(batch_df, batch_id):
    def write_partition(rows):
        # Open one client per partition.
        for row in rows:
            # Use an idempotency key (order_id) so a retried batch doesn't double-write.
            upsert_to_external_system(key=row.order_id, payload=row.asDict())

    batch_df.select("order_id", "amount").foreachPartition(write_partition)

For more information about writing to external systems, see Sinks in Lakeflow pipelines.

Deduplicate at-least-once sources

When a source can deliver a record more than once, deduplicate downstream. Combine a watermark with dropDuplicatesWithinWatermark, which is watermark-aware and doesn't require unbounded state to detect duplicates. Deduplicate on the columns that uniquely identify an event. The identity can span several columns when no single column is unique on its own. In the following example, a click sequence number is unique only within its session, so the two columns together identify the event:

from pyspark import pipelines as dp

@dp.table(name="clicks_deduped")
def clicks_deduped():
    return (
        spark.readStream.table("clicks_bronze")
        .withWatermark("click_ts", "5 minutes")
        .dropDuplicatesWithinWatermark(["session_id", "click_seq_num"])
    )

Choose those columns from the source's uniqueness contract, not from what looks distinct in sample data. Columns that can legitimately repeat discard real events when you treat them as the identity. A user clicking the same ad twice is a common example: deduplicating on the user and the ad silently drops the second click.

AUTO CDC's key-based upsert semantics also collapse duplicates naturally, so routing at-least-once data through an AUTO CDC flow keyed on a stable business key is another way to converge on exactly-once state.

Limitations

Exactly-once processing applies to managed Delta-to-Delta flows. Treat the following edges as at-least-once and add explicit deduplication or idempotent-write logic there:

  • foreach_batch_sink and custom external writes. Spark guarantees a batch is attempted at least once, but a batch retried after a partial write can leave some rows visible twice in the external system. Make the external write idempotent, for example by upserting on a natural key or writing a batch ID the receiver can deduplicate on.
  • Kafka as a sink. Kafka topics don't support transactional exactly-once writes the way Delta does, so a retried micro-batch writing to Kafka can produce duplicate messages. If downstream consumers are sensitive to duplicates, dedupe on the consumer side, for example by event ID.
  • Custom Python data sources used as sources. Whether reads are exactly-once depends on whether your source implementation correctly reports and resumes from offsets. If it doesn't track offsets, treat it as at-least-once and dedupe downstream with dropDuplicates on an event ID or by relying on AUTO CDC's key-based upsert semantics.

As a rule of thumb, if your entire pipeline is Delta-to-Delta (streaming tables and materialized views reading and writing Delta tables through managed flows), you already have exactly-once. The moment you add a foreach_batch_sink, a non-Delta sink, or an unverified custom source, treat that specific edge as at-least-once and add idempotent-write or deduplication logic there.

Additional resources

  • Last updated on