# Build an incremental load into a Delta table with MERGE, safe to rerun An incremental load is safe to rerun when every write it makes either replaces its own earlier output or changes only what is newer. load_orders.py does both for the shop's orders: it replaces the rejected lines of the file it loads with replaceWhere, keeps the newest change per order, and merges with a condition on updated_at. Loading 2 days, then the second day again, then the first day again leaves the same 5 orders. - Page: https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/build-it - Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 4: Tables: Parquet, Delta Lake and Iceberg, Build it: an incremental load - Parts of this module: The transaction log and time travel (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg); MERGE (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/merge); Schema enforcement and evolution (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/schema); OPTIMIZE, clustering and VACUUM (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/optimize-and-vacuum); Iceberg and UniForm (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/iceberg-and-uniform); Build it: an incremental load (https://tecyfy.com/data-engineering/course/delta-lake-and-iceberg/build-it) At the end of this part you can: - Write an incremental load that merges a day's changes into a Delta table - Make every write of a job safe to repeat, with a conditional MERGE and replaceWhere - Stop a load when its input's columns change, before MERGE drops a column The course's order pipeline so far cleans a raw file into Parquet folders. A real shop sends a new file of changes every day, and the table must take each one in: new orders, changed orders, late or repeated rows, now and then a broken line. This part builds that step on a Delta table, from parts 1 to 4. ## Build it: load a day's changes First make 2 days of files. The first is the raw file from module 2, with its repeated rows and the line that does not parse. Save this as `make_changes.py`: ```python from pathlib import Path RAW = Path("raw/changes") RAW.mkdir(parents=True, exist_ok=True) HEADER = "order_id,customer_id,status,amount,currency,ordered_at,updated_at\n" (RAW / "2026-10-03.csv").write_text(HEADER + """1001,C01,placed,25.50,EUR,2026-10-01T09:15:00,2026-10-01T09:15:00 1002,C02,placed,12.00,EUR,2026-10-01T10:02:00,2026-10-01T10:02:00 1001,C01,shipped,25.50,EUR,2026-10-01T09:15:00,2026-10-02T08:00:00 1003,C03,placed,abc,EUR,2026-10-02T11:30:00,2026-10-02T11:30:00 1004,C01,placed,40.00,EUR,2026-10-02T12:45:00,2026-10-02T12:45:00 1002,C02,cancelled,12.00,EUR,2026-10-01T10:02:00,2026-10-03T09:00:00 1005,C09,placed,8.75,EUR,2026-10-03T14:10:00,2026-10-03T14:10:00 1004,C01,placed,40.00,EUR,2026-10-02T12:45:00,2026-10-02T12:45:00 """) (RAW / "2026-10-04.csv").write_text(HEADER + """1002,C02,placed,12.00,EUR,2026-10-01T10:02:00,2026-10-01T10:02:00 1004,C01,shipped,40.00,EUR,2026-10-02T12:45:00,2026-10-04T08:30:00 1005,C09,cancelled,8.75,EUR,2026-10-03T14:10:00,2026-10-04T10:00:00 1006,C02,placed,19.90,EUR,2026-10-04T11:20:00,2026-10-04T11:20:00 1006,C02,shipped,19.90,EUR,2026-10-04T11:20:00,2026-10-04T16:45:00 """) ``` The second day ships 1004, cancels 1005 and adds 1006, which is placed and shipped on the same day. It also resends an old row for 1002. Save this as `load_orders.py`, in a new folder with `delta_session.py` from [part 1](/data-engineering/course/delta-lake-and-iceberg): ```python """Load 1 file of order changes into the orders table. Safe to run again on the same file.""" import sys from delta import DeltaTable from pyspark.sql import DataFrame, SparkSession, Window from pyspark.sql import functions as F ORDERS = """ order_id BIGINT, customer_id STRING, status STRING, amount DECIMAL(10,2), currency STRING, ordered_at TIMESTAMP, updated_at TIMESTAMP """ def check_columns(spark: SparkSession, path: str) -> None: """Stop if the file's header differs from the columns the job expects.""" found = spark.read.option("header", True).csv(path).columns expected = spark.createDataFrame([], ORDERS).columns if found != expected: raise RuntimeError(f"{path} has columns {found}, expected {expected}") def read_changes(spark: SparkSession, path: str) -> DataFrame: """Every row of the file, with the lines that do not parse kept in _bad.""" return ( spark.read.schema(ORDERS + ", _bad STRING") .option("header", True) .option("columnNameOfCorruptRecord", "_bad") .csv(path) .withColumn("source_file", F.lit(path)) ) def newest_per_order(changes: DataFrame) -> DataFrame: """1 row per order, its newest change, so MERGE never sees 2 rows for 1 order.""" newest_first = Window.partitionBy("order_id").orderBy(F.col("updated_at").desc()) return changes.withColumn("n", F.row_number().over(newest_first)).filter("n = 1").drop("n") def merge_orders(spark: SparkSession, latest: DataFrame) -> DataFrame: """Update an order only with a newer change, and add the orders not seen before.""" spark.sql(f"CREATE TABLE IF NOT EXISTS orders ({ORDERS}) USING DELTA") return ( DeltaTable.forName(spark, "orders").alias("t") .merge(latest.alias("s"), "t.order_id = s.order_id") .whenMatchedUpdateAll(condition="s.updated_at > t.updated_at") .whenNotMatchedInsertAll() .execute() ) def main(spark: SparkSession, path: str) -> None: check_columns(spark, path) rows = read_changes(spark, path) # This file's rejected lines replace any from an earlier run of the same file. ( rows.filter(F.col("_bad").isNotNull()) .write.format("delta") .mode("overwrite") .option("replaceWhere", f"source_file = '{path}'") .saveAsTable("rejected_orders") ) rejected = spark.table("rejected_orders").filter(F.col("source_file") == path).count() good = rows.filter(F.col("_bad").isNull()).drop("_bad", "source_file") result = merge_orders(spark, newest_per_order(good)).first() # Check: the table must still hold 1 row per order. orders = spark.table("orders") if orders.count() != orders.select("order_id").distinct().count(): raise RuntimeError("an order appears twice in the orders table") version = spark.sql("DESCRIBE HISTORY orders LIMIT 1").first().version print( f"{path}: {rejected} rejected, {result.num_updated_rows} updated, " f"{result.num_inserted_rows} added, table version {version}" ) if __name__ == "__main__": from delta_session import delta_spark # on Databricks, use the spark it gives you main(delta_spark("load-orders"), sys.argv[1]) ``` Load the first day, the second, then the second again and the first again, as a confused scheduler might: ```bash python make_changes.py python load_orders.py raw/changes/2026-10-03.csv python load_orders.py raw/changes/2026-10-04.csv python load_orders.py raw/changes/2026-10-04.csv python load_orders.py raw/changes/2026-10-03.csv ``` ```text raw/changes/2026-10-03.csv: 1 rejected, 0 updated, 4 added, table version 1 raw/changes/2026-10-04.csv: 0 rejected, 2 updated, 1 added, table version 2 raw/changes/2026-10-04.csv: 0 rejected, 0 updated, 0 added, table version 3 raw/changes/2026-10-03.csv: 1 rejected, 0 updated, 0 added, table version 4 ``` The table ends with 5 orders, each in its latest state, and 1 rejected line. Save this as `show.py` to see them: ```python from delta_session import delta_spark spark = delta_spark() spark.sql("SELECT order_id, status, amount, updated_at FROM orders ORDER BY order_id").show() spark.sql("SELECT order_id, _bad, source_file FROM rejected_orders").show(truncate=False) ``` ```text +--------+---------+------+-------------------+ |order_id| status|amount| updated_at| +--------+---------+------+-------------------+ | 1001| shipped| 25.50|2026-10-02 08:00:00| | 1002|cancelled| 12.00|2026-10-03 09:00:00| | 1004| shipped| 40.00|2026-10-04 08:30:00| | 1005|cancelled| 8.75|2026-10-04 10:00:00| | 1006| shipped| 19.90|2026-10-04 16:45:00| +--------+---------+------+-------------------+ +--------+---------------------------------------------------------------+--------------------------+ |order_id|_bad |source_file | +--------+---------------------------------------------------------------+--------------------------+ |1003 |1003,C03,placed,abc,EUR,2026-10-02T11:30:00,2026-10-02T11:30:00|raw/changes/2026-10-03.csv| +--------+---------------------------------------------------------------+--------------------------+ ``` ## What makes it safe to rerun The reruns changed nothing, and that is the point of the design: when a job fails halfway or runs twice, you run it again and need not think. A job is only as safe as its least careful write, so each of its writes is built for it. **The MERGE changes only what is newer.** The condition on `updated_at` leaves the repeated and late rows alone, so the second day's old 1002 and both reruns updated nothing. Keeping the newest row per order first, `newest_per_order`, means 1006's 2 rows become 1 shipped order, with no error and no duplicate. **The other write replaces its own rows.** The rejected lines go in with `replaceWhere` on the file they came from, so a rerun replaces them instead of adding a second copy, and other files' lines stay. A plain append there would count the same bad line again on every run. **The input is checked before anything is written.** `check_columns` compares the file's header with the job's columns and stops on any difference, a new column, a missing one or the same columns in another order. [Part 3](/data-engineering/course/delta-lake-and-iceberg/schema) showed why: MERGE would drop a new column without an error, and the CSV reader applies its schema by position, so a column added in the middle would shift every value after it. Each run that changes nothing still makes a commit, versions 3 and 4 here. That is harmless, and the history shows every run. ## On Databricks Put the 2 files in a volume, as in [module 2](/data-engineering/course/pyspark-dataframes), and pass the volume path, such as `/Volumes///raw/changes/2026-10-03.csv`. In a notebook, call `main(spark, path)` with the session Databricks gives you, and leave out the `if __name__` block. Drop `USING DELTA` if you like, since Delta is the default, and the tables land in your current catalog and schema. Then try 2 changes on your own. Add a `coupon` column to a copy of the second day's file and load it: the run stops on the header check and names both lists of columns. Then delete the `condition=` from the MERGE, load the first day after the second, and watch 1004 and 1005 go back to placed. Next, [module 5](/data-engineering/course/unity-catalog) puts these tables in Unity Catalog and decides who can read them. - Quiz: [Delta Lake Essentials](https://tecyfy.com/quiz/data-engineering/delta-lake-essentials). Test what you know about Delta Lake tables: the transaction log, concurrent writes, time travel and VACUUM, MERGE, schema enforcement and evolution, the change… ## Check yourself 1. load_orders.py writes the rejected lines with mode("overwrite") and replaceWhere on source_file. What would go wrong with a plain append instead? A. Nothing, since the rejected table is only for people to read B. Each rerun of the same file would add its rejected lines again, so the rejected table would count 1 bad line as 2, 3 or more C. The append would fail, because the table already exists D. The append would delete the rejected lines of other files ### Answers 1. B. A job that may run twice must make every write repeatable, not just the main one. An append adds the same rows each time. replaceWhere replaces only the rows that match its condition, here the lines from this file, in 1 commit, so a rerun leaves 1 copy and the other files' lines are untouched. ## Questions ### What is an idempotent data load? A load is idempotent when running it again with the same input leaves the same result as running it once. A load that MERGEs changes with a condition such as s.updated_at > t.updated_at, and replaces its other outputs instead of appending to them, is idempotent, so a failed or repeated run can simply be run again. ### What does replaceWhere do in Delta Lake? replaceWhere is a write option for mode("overwrite") that replaces only the rows matching a condition, such as source_file = 'raw/changes/2026-10-03.csv', and leaves the rest of the table alone, in 1 commit. Delta checks that the data written matches the condition too. It suits a job that rewrites its own share of a table. ### How do I load files incrementally into a Delta table? Read only the new file or files, keep 1 row per key, its newest change, and MERGE it into the table on the key, updating only with newer changes and inserting new keys. On Databricks, Auto Loader finds the new files for you and remembers which it has loaded, which module 6 covers. ## Sources - [Selectively overwrite data with Delta Lake](https://docs.databricks.com/aws/en/delta/selective-overwrite) - [Table batch reads and writes, overwrite with replaceWhere](https://docs.delta.io/latest/delta-batch.html) - [Table deletes, updates, and merges](https://docs.delta.io/latest/delta-update.html) - [CSV files, options](https://spark.apache.org/docs/latest/sql-data-sources-csv.html)