Making Batch Pipelines Retry-Safe with MERGE INTO

[12]of 16

Ollie

head gardener

In Parts 8 through 10, we wrote derived rows with INSERT INTO in Spark SQL and .writeTo().append() in the DataFrame API.

That works well when every data interval is processed exactly once. Unfortunately, jobs do not always live such peaceful lives.

A job may finish writing its rows and then appear to fail. An operator may retry it, or a scheduler may retry it automatically. If the job appends the same interval again, the destination table receives duplicate rows.

In this part, we will return to the cart-abandonment job from Part 10 and replace its append operation with Iceberg's MERGE INTO statement.

Our new jobs will make the destination match the recomputed result for one requested interval. This makes that interval safer to retry, but it does not turn our tutorial into a complete production pipeline. Concurrent writers, commit-conflict recovery, and orchestration policies are still waiting patiently outside the door.

How MERGE INTO Works

MERGE INTO compares a source dataset with a target table.

The target is the table that we want to change. The source contains the rows that describe what the target should become.

The basic shape looks like this:

sql

MERGE INTO target_table AS target
USING (
  SELECT ...
) AS source
  ON target.key = source.key
WHEN MATCHED THEN ...
WHEN NOT MATCHED THEN ...
WHEN NOT MATCHED BY SOURCE THEN ...

MERGE INTO and USING introduce the target and source. The aliases target and source make it clear which dataset a column comes from.

The ON condition defines how Spark decides that a source row and a target row represent the same thing. It works much like a join condition, but its purpose here is to find the target row that a source row may change.

It does not filter the source to the requested date range. We still do that inside the source query, just as we did in the earlier jobs.

After applying the ON condition, every relevant row falls into one of three groups:

Relationship	Meaning
Matched	A corresponding row exists in both the source and target
Not matched	A source row has no corresponding target row
Not matched by source	A target row has no corresponding source row

The WHEN clauses tell Spark what to do with each group.

Understanding the Three WHEN Clauses

Each match state allows particular actions:

Clause	Row relationship	Allowed actions
`WHEN MATCHED`	Source and target both contain the row	`UPDATE` or `DELETE`
`WHEN NOT MATCHED`	Only the source contains the row	`INSERT`
`WHEN NOT MATCHED BY SOURCE`	Only the target contains the row	`UPDATE` or `DELETE`

Let’s look at them one at a time.

WHEN MATCHED

WHEN MATCHED handles a source row that found a target row through the ON condition.

We can update that target row:

sql

WHEN MATCHED THEN UPDATE SET
  target.metric = source.metric

Or we can delete it:

sql

WHEN MATCHED THEN DELETE

An optional condition can make the action more specific:

sql

WHEN MATCHED AND source.should_delete THEN DELETE

This is useful when different matched rows require different actions.

WHEN NOT MATCHED

WHEN NOT MATCHED handles a source row that did not find a target row.

Because the target row does not exist yet, the available action is INSERT:

sql

WHEN NOT MATCHED THEN INSERT (key, metric)
VALUES (source.key, source.metric)

This clause can also have a condition:

sql

WHEN NOT MATCHED AND source.metric > 0 THEN INSERT (key, metric)
VALUES (source.key, source.metric)

WHEN NOT MATCHED BY SOURCE

WHEN NOT MATCHED BY SOURCE looks in the opposite direction. It handles a target row that did not find a source row.

We can update that target row:

sql

WHEN NOT MATCHED BY SOURCE THEN UPDATE SET
  target.is_active = FALSE

Or delete it:

sql

WHEN NOT MATCHED BY SOURCE THEN DELETE

Spark 3.5 added this clause. It is especially useful when the source represents a complete replacement for a particular scope and target rows missing from that source should no longer remain.

Conditions matter when more than one clause could apply. Spark evaluates eligible clauses in order and uses the first applicable action. A row for which no action applies remains unchanged.

There is one more important rule: no more than one source row may update the same target row. Otherwise, Spark cannot know which source row should win, and Iceberg rejects the merge.

Our source groups rows by event_date, product_id, product_name, and category. Those same columns form the merge condition, so the source produces at most one row for each target key.

MERGE INTO works in this tutorial because the target is an Iceberg table and Spark is configured with the Iceberg SQL extensions. Row-level operations depend on the table provider. Another provider or version may not support MERGE INTO, or it may not support every clause and action shown here. The Iceberg Spark write documentation describes the supported Iceberg behavior.

Upsert Versus Complete Reconciliation

A common use of MERGE INTO is an upsert: update rows that already exist and insert rows that are new.

For our cart-abandonment table, the action portion of an upsert could look like this:

sql

WHEN MATCHED THEN UPDATE SET
  target.cart_session_count = source.cart_session_count,
  target.purchased_cart_session_count = source.purchased_cart_session_count,
  target.abandoned_cart_session_count = source.abandoned_cart_session_count,
  target.cart_abandonment_rate = source.cart_abandonment_rate
WHEN NOT MATCHED THEN INSERT (
  event_date,
  product_id,
  product_name,
  category,
  cart_session_count,
  purchased_cart_session_count,
  abandoned_cart_session_count,
  cart_abandonment_rate
) VALUES (
  source.event_date,
  source.product_id,
  source.product_name,
  source.category,
  source.cart_session_count,
  source.purchased_cart_session_count,
  source.abandoned_cart_session_count,
  source.cart_abandonment_rate
)

This is enough to retry the same unchanged input without appending another copy. However, it leaves target-only rows untouched.

Imagine that a corrected source no longer contains any cart activity for one product. That product will disappear from the new aggregate source. An upsert does not remove its old destination row, so the target no longer reflects the complete recomputed interval.

Our Part 12 jobs perform complete reconciliation for the interval. They add:

sql

WHEN NOT MATCHED BY SOURCE
  AND target.event_date >= CAST(CAST('{data_interval_start}' AS TIMESTAMP) AS DATE)
  AND target.event_date < CAST(CAST('{data_interval_end}' AS TIMESTAMP) AS DATE)
  THEN DELETE

The date conditions are not decorative. The source query contains only the requested interval, so every target row outside that interval is also absent from the source. An unbounded WHEN NOT MATCHED BY SOURCE THEN DELETE could therefore delete the rest of the table.

Reconciliation is appropriate only when the source represents the complete truth for its scope. If the source contains only incremental changes, a missing row does not necessarily mean that the target row should be deleted. In that case, use an upsert or define a different deletion signal.

Reusing the Part 10 Tables and Transformation

Part 12 reuses the source and destination tables from Part 10.

The SQL job reads retail_session_products and merges into retail_daily_product_cart_abandonment. The DataFrame job uses the equivalent tables with the _df suffix.

The destination tables must already exist. Refer to Part 10 for their DDL; the jobs still transform data rather than create tables.

Initialize the SQL project:

bash

oleander spark init part12-sql

Use:

tutorial_part12_sql.py as the entrypoint file name
tutorial as the custom library name

Then initialize the DataFrame project:

bash

oleander spark init part12-dataframe

Use:

tutorial_part12_dataframe.py as the entrypoint file name
tutorial as the custom library name

Run uv sync in each project after initialization.

Both jobs keep the familiar required arguments, UTC timezone, and half-open date interval. They also keep Part 10's cart and purchase filters, left join, and product-level aggregation.

The important change is what happens after that aggregation is ready.

Merging with Spark SQL

Here is the complete SQL transformation:

python

def build_daily_product_cart_abandonment(
    spark: SparkSession,
    data_interval_start: str,
    data_interval_end: str,
) -> None:
    """
    Build daily product cart-abandonment rows for the given data interval.

    This function joins cart sessions with purchase sessions, summarizes the
    outcomes by event date and product, and merges the result into
    oleander.tutorial.retail_daily_product_cart_abandonment.

    The output table is expected to already exist.
    """
    spark.sql(f"""
        MERGE INTO oleander.tutorial.retail_daily_product_cart_abandonment
          AS target
        USING (
          WITH filtered_session_products AS (
            SELECT *
            FROM oleander.tutorial.retail_session_products
            WHERE event_date >= CAST(CAST('{data_interval_start}' AS TIMESTAMP) AS DATE)
              AND event_date < CAST(CAST('{data_interval_end}' AS TIMESTAMP) AS DATE)
          ),
          cart_sessions AS (
            SELECT
              session_id,
              event_date,
              product_id,
              product_name,
              category
            FROM filtered_session_products
            WHERE added_to_cart
          ),
          purchase_sessions AS (
            SELECT
              session_id,
              product_id,
              TRUE AS purchased_cart
            FROM filtered_session_products
            WHERE purchased
          ),
          cart_outcomes AS (
            SELECT
              cart.event_date,
              cart.product_id,
              cart.product_name,
              cart.category,
              COALESCE(purchase.purchased_cart, FALSE) AS purchased_cart
            FROM cart_sessions AS cart
            LEFT JOIN purchase_sessions AS purchase
              ON cart.session_id = purchase.session_id
             AND cart.product_id = purchase.product_id
          ),
          cart_summary AS (
            SELECT
              event_date,
              product_id,
              product_name,
              category,
              COUNT(*) AS cart_session_count,
              SUM(CASE WHEN purchased_cart THEN 1 ELSE 0 END)
                AS purchased_cart_session_count,
              SUM(CASE WHEN NOT purchased_cart THEN 1 ELSE 0 END)
                AS abandoned_cart_session_count
            FROM cart_outcomes
            GROUP BY
              event_date,
              product_id,
              product_name,
              category
          )
          SELECT
            event_date,
            product_id,
            product_name,
            category,
            cart_session_count,
            purchased_cart_session_count,
            abandoned_cart_session_count,
            CAST(
              CASE
                WHEN cart_session_count > 0
                  THEN abandoned_cart_session_count / cart_session_count
                ELSE 0
              END AS DECIMAL(9, 4)
            ) AS cart_abandonment_rate
          FROM cart_summary
        ) AS source
          ON target.event_date = source.event_date
         AND target.product_id = source.product_id
         AND target.product_name = source.product_name
         AND target.category = source.category
        WHEN MATCHED THEN UPDATE SET
          target.cart_session_count = source.cart_session_count,
          target.purchased_cart_session_count = source.purchased_cart_session_count,
          target.abandoned_cart_session_count = source.abandoned_cart_session_count,
          target.cart_abandonment_rate = source.cart_abandonment_rate
        WHEN NOT MATCHED THEN INSERT (
          event_date,
          product_id,
          product_name,
          category,
          cart_session_count,
          purchased_cart_session_count,
          abandoned_cart_session_count,
          cart_abandonment_rate
        ) VALUES (
          source.event_date,
          source.product_id,
          source.product_name,
          source.category,
          source.cart_session_count,
          source.purchased_cart_session_count,
          source.abandoned_cart_session_count,
          source.cart_abandonment_rate
        )
        WHEN NOT MATCHED BY SOURCE
          AND target.event_date >= CAST(CAST('{data_interval_start}' AS TIMESTAMP) AS DATE)
          AND target.event_date < CAST(CAST('{data_interval_end}' AS TIMESTAMP) AS DATE)
          THEN DELETE
    """)

The CTEs are the same transformation stages from Part 10. This time, their final SELECT sits inside USING (...) and becomes the merge source.

The ON condition uses all four columns that define the destination grain:

sql

ON target.event_date = source.event_date
AND target.product_id = source.product_id
AND target.product_name = source.product_name
AND target.category = source.category

When that identity already exists, the job updates its metrics. The identity columns are not updated because they are the columns that established the match.

When the identity does not exist, the job inserts the complete row. We list every inserted column and value explicitly so that the mapping remains visible.

Finally, a target row missing from the source is deleted only if its date belongs to the requested interval. Together, those actions make the target interval agree with the newly computed source interval.

Merging a DataFrame Result

The DataFrame transformation still uses .filter(), .join(), .groupBy(), and .agg() to build cart_summary.

The ending is different from the earlier DataFrame jobs. This tutorial uses Spark 3.5.5, which does not provide a DataFrame-native merge writer. Spark 4 introduced one, but changing Spark versions would be a rather large footnote for one method call.

Instead, we register the final DataFrame as a temporary view and use it from Spark SQL.

Here is the complete transformation:

python

from pyspark.sql import functions as F


def build_daily_product_cart_abandonment(
    spark: SparkSession,
    data_interval_start: str,
    data_interval_end: str,
) -> None:
    """
    Build daily product cart-abandonment rows for the given data interval.

    This function joins cart sessions with purchase sessions, summarizes the
    outcomes by event date and product, and merges the result into
    oleander.tutorial.retail_daily_product_cart_abandonment_df.

    The output table is expected to already exist.
    """
    start_date = F.lit(data_interval_start).cast("timestamp").cast("date")
    end_date = F.lit(data_interval_end).cast("timestamp").cast("date")

    session_products = (
        spark.table("oleander.tutorial.retail_session_products_df")
        .filter(F.col("event_date") >= start_date)
        .filter(F.col("event_date") < end_date)
    )

    cart_sessions = (
        session_products
        .filter(F.col("added_to_cart"))
        .select(
            "session_id",
            "event_date",
            "product_id",
            "product_name",
            "category",
        )
    )

    purchase_sessions = (
        session_products
        .filter(F.col("purchased"))
        .select(
            "session_id",
            "product_id",
            F.lit(True).alias("purchased_cart"),
        )
    )

    cart_outcomes = (
        cart_sessions
        .join(
            purchase_sessions,
            ["session_id", "product_id"],
            "left",
        )
        .withColumn(
            "purchased_cart",
            F.coalesce(F.col("purchased_cart"), F.lit(False)),
        )
    )

    cart_summary = (
        cart_outcomes
        .groupBy(
            "event_date",
            "product_id",
            "product_name",
            "category",
        )
        .agg(
            F.count(F.lit(1)).alias("cart_session_count"),
            F.sum(
                F.when(F.col("purchased_cart"), F.lit(1))
                .otherwise(F.lit(0))
            ).alias("purchased_cart_session_count"),
            F.sum(
                F.when(~F.col("purchased_cart"), F.lit(1))
                .otherwise(F.lit(0))
            ).alias("abandoned_cart_session_count"),
        )
        .withColumn(
            "cart_abandonment_rate",
            F.when(
                F.col("cart_session_count") > F.lit(0),
                F.col("abandoned_cart_session_count")
                / F.col("cart_session_count"),
            )
            .otherwise(F.lit(0))
            .cast("decimal(9,4)"),
        )
        .select(
            "event_date",
            "product_id",
            "product_name",
            "category",
            "cart_session_count",
            "purchased_cart_session_count",
            "abandoned_cart_session_count",
            "cart_abandonment_rate",
        )
    )

    source_view = "daily_product_cart_abandonment_updates"
    cart_summary.createOrReplaceTempView(source_view)

    try:
        spark.sql(f"""
            MERGE INTO oleander.tutorial.retail_daily_product_cart_abandonment_df
              AS target
            USING {source_view} AS source
              ON target.event_date = source.event_date
             AND target.product_id = source.product_id
             AND target.product_name = source.product_name
             AND target.category = source.category
            WHEN MATCHED THEN UPDATE SET
              target.cart_session_count = source.cart_session_count,
              target.purchased_cart_session_count = source.purchased_cart_session_count,
              target.abandoned_cart_session_count = source.abandoned_cart_session_count,
              target.cart_abandonment_rate = source.cart_abandonment_rate
            WHEN NOT MATCHED THEN INSERT (
              event_date,
              product_id,
              product_name,
              category,
              cart_session_count,
              purchased_cart_session_count,
              abandoned_cart_session_count,
              cart_abandonment_rate
            ) VALUES (
              source.event_date,
              source.product_id,
              source.product_name,
              source.category,
              source.cart_session_count,
              source.purchased_cart_session_count,
              source.abandoned_cart_session_count,
              source.cart_abandonment_rate
            )
            WHEN NOT MATCHED BY SOURCE
              AND target.event_date >= CAST(CAST('{data_interval_start}' AS TIMESTAMP) AS DATE)
              AND target.event_date < CAST(CAST('{data_interval_end}' AS TIMESTAMP) AS DATE)
              THEN DELETE
        """)
    finally:
        spark.catalog.dropTempView(source_view)

createOrReplaceTempView() gives cart_summary a SQL name for the current Spark session. It does not copy the rows into another permanent table.

The SQL statement can then use that name in its USING clause:

sql

USING daily_product_cart_abandonment_updates AS source

The merge actions are otherwise the same as the SQL job.

The finally block removes the temporary view whether the merge succeeds or raises an error. The Spark session would remove it when the job stops anyway, but explicit cleanup makes ownership clear and keeps the function safer to reuse in a longer session.

Verifying Retries and Reconciliation

Run either job for a clean interval, then run the same job again with the same arguments.

After the second run, check for duplicate keys. For the SQL destination, use:

python

spark.sql("""
    SELECT
      event_date,
      product_id,
      product_name,
      category,
      COUNT(*) AS row_count
    FROM oleander.tutorial.retail_daily_product_cart_abandonment
    GROUP BY
      event_date,
      product_id,
      product_name,
      category
    HAVING COUNT(*) > 1
    ORDER BY
      event_date,
      product_id,
      product_name,
      category
""").show(truncate=False)

Use retail_daily_product_cart_abandonment_df to check the DataFrame destination.

For a destination that began with one row per key, the query should return no rows. The retry updates matching rows instead of appending another copy.

If the recomputed metrics change, WHEN MATCHED replaces the old metric values. If a new product appears, WHEN NOT MATCHED inserts it. If a product disappears from the complete source interval, the bounded WHEN NOT MATCHED BY SOURCE clause deletes its old row. Target rows on other dates remain untouched.

There are two practical caveats.

First, this merge does not repair duplicates that already existed before Part 12. If two target rows have the same key, matching them again does not magically turn them into one row. Start the verification with an interval that has no existing duplicates.

Second, our unconditional WHEN MATCHED action may update an identical row. Iceberg implements row-level changes by rewriting affected data files, so a retry can still perform physical work even when the logical values do not change. Avoiding unchanged updates is a useful optimization, but it would distract from the merge syntax we are learning here.

Full Example Code

The complete Part 12 jobs are available here:

Deploying and Submitting the Jobs

The deployment process remains the same as in the earlier parts.

Build the artifacts from each project:

bash

make

Upload the SQL job:

bash

oleander spark jobs upload tutorial_part12_sql.py

Or upload the DataFrame job:

bash

oleander spark jobs upload tutorial_part12_dataframe.py

Submit the SQL version for one day:

bash

oleander spark jobs submit tutorial_part12_sql.py \
  --namespace tutorial \
  --name tutorial_part12_sql \
  --args=--data-interval-start=2026-01-01T00:00:00Z \
  --args=--data-interval-end=2026-01-02T00:00:00Z

Or submit the DataFrame version:

bash

oleander spark jobs submit tutorial_part12_dataframe.py \
  --namespace tutorial \
  --name tutorial_part12_dataframe \
  --args=--data-interval-start=2026-01-01T00:00:00Z \
  --args=--data-interval-end=2026-01-02T00:00:00Z

Submitting the same job again with the same interval should leave the destination with the same logical rows rather than append duplicates.

Summary

In this part, we replaced the append-only write from Part 10 with MERGE INTO.

We learned that WHEN MATCHED updates or deletes rows found on both sides, WHEN NOT MATCHED inserts new source rows, and WHEN NOT MATCHED BY SOURCE updates or deletes target-only rows.

Using the first two clauses creates an upsert. Adding a carefully bounded target-only deletion lets our jobs completely reconcile the requested interval.

The SQL job uses the aggregate query directly as its merge source. The DataFrame job creates the same aggregate with the DataFrame API, registers it as a temporary view, and performs the Iceberg row-level operation with Spark SQL.

Our jobs are now safer to retry, but retries may still rewrite data files and preexisting duplicates remain preexisting duplicates. Production pipelines always keep a little humility in reserve.

In the next part, we will turn from correctness to query performance and use Iceberg partitioning to help Spark skip data that a date-filtered query does not need.

PreviousUnderstanding Join Types in Spark NextImproving Query Performance with Partitioning