# Making Batch Pipelines Retry-Safe with MERGE INTO > Growing Tips (Apache Iceberg & Spark 101), part 12 of 16. > Retries happen. MERGE INTO keeps a re-run pipeline from double-planting the same rows. > Source: https://oleander.dev/grow/making-batch-pipelines-retry-safe-with-merge-into 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](https://iceberg.apache.org/docs/latest/spark-writes/) 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: - [SQL version](https://github.com/OleanderHQ/iceberg-spark-101-tutorials/blob/main/part12-sql/tutorial_part12_sql.py) - [DataFrame version](https://github.com/OleanderHQ/iceberg-spark-101-tutorials/blob/main/part12-dataframe/tutorial_part12_dataframe.py) ## 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.