Only the new rows,every run.

Merging all of bronze into silver on every run works while bronze is small. As it grows, each run reads more rows to change fewer. A watermark fixes that: it remembers how far the last good run got, so the next run reads only what came after.

#What to measure by

The watermark needs a column that only goes up as new rows arrive. In the bronze table from the lakehouse course, every row has _loaded_at, the time it was landed. That is the column to use: it belongs to the platform, not to the source, so a source that sends old dates cannot fool it.

#The table

In a notebook attached to the lakehouse, create one Delta table for every watermark the lakehouse will ever need:

spark.sql("""
    CREATE TABLE IF NOT EXISTS etl_watermark (
        table_name  STRING,
        last_value  TIMESTAMP,
        updated_at  TIMESTAMP
    ) USING DELTA
""")

One row per table you load this way. Add the row for orders, starting far in the past so the first run reads everything:

from pyspark.sql import functions as F

if spark.table("etl_watermark").filter("table_name = 'orders'").isEmpty():
    spark.createDataFrame(
        [("orders",)], "table_name STRING"
    ).select(
        "table_name",
        F.lit("1900-01-01").cast("timestamp").alias("last_value"),
        F.current_timestamp().alias("updated_at"),
    ).write.mode("append").saveAsTable("etl_watermark")

The check before the insert means the cell can run again without adding a second row.

#Read it

Every run starts by asking how far the last one got:

last = (spark.table("etl_watermark")
        .filter("table_name = 'orders'")
        .first()["last_value"])

print(f"Reading orders loaded after {last}")

Run it now: it prints the start date, 1 January 1900.

#Why a table and not a file

A Delta table can be read and changed by any notebook, pipeline or SQL query in the workspace, keeps its history and is changed in one step or not at all. You can also look at it from the lakehouse's SQL analytics endpoint:

SELECT table_name, last_value, updated_at
FROM etl_watermark;

With the watermark read, the notebook takes only the bronze rows loaded after it, merges them into silver by key, and only then moves the watermark on. If any step fails, the watermark stays where it was and the next run reads the same rows again.

#The new rows

from pyspark.sql import functions as F, Window

new = spark.table("bronze_orders").filter(F.col("_loaded_at") > F.lit(last))

if new.isEmpty():
    notebookutils.notebook.exit("0")

notebookutils is built into every Fabric notebook. Exiting early when there is nothing new keeps a quiet night quick.

#Clean them as silver expects

The rules from the lakehouse course still apply to the new rows: types, trimmed text and one row per order, newest first. Keep them in one place, a function that gives silver every one of its columns:

def clean(df):
    typed = df.select(
        F.trim("OrderID").alias("order_id"),
        F.to_date("OrderDate", "yyyy-MM-dd").alias("order_date"),
        F.trim("CustomerID").alias("customer_id"),
        F.trim("CustomerName").alias("customer_name"),
        F.trim("Country").alias("country"),
        F.trim("Product").alias("product"),
        F.col("Quantity").cast("int").alias("quantity"),
        F.col("Amount").cast("decimal(18,2)").alias("amount"),
        "_loaded_at",
    ).filter("order_id IS NOT NULL AND order_id <> ''")
    newest = Window.partitionBy("order_id").orderBy(F.col("_loaded_at").desc())
    return typed.withColumn("_n", F.row_number().over(newest)).filter("_n = 1").drop("_n")

rows = clean(new)

#Merge them into silver

The quarantine of the lakehouse course belongs here too; it is left out to keep the merge short.

from delta.tables import DeltaTable

(DeltaTable.forName(spark, "silver_orders").alias("t")
    .merge(rows.alias("s"), "t.order_id = s.order_id")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

Only the new rows are read and merged, however big silver has grown.

#Move the watermark on

The new watermark is the latest load time among the rows just merged, not the clock. A row landed while the notebook ran then still counts as new next time.

high = new.agg(F.max("_loaded_at")).first()[0]

DeltaTable.forName(spark, "etl_watermark").update(
    condition="table_name = 'orders'",
    set={"last_value": F.lit(high), "updated_at": F.current_timestamp()},
)

notebookutils.notebook.exit(str(rows.count()))

This cell is last on purpose. A failed merge raises an error and stops the notebook before it is reached.

#Try it

Run the notebook twice. The first run merges every order and moves the watermark to the latest load; the second exits at once with 0. Land one more file in bronze, run again, and only its rows are read.

The notebook works when you press Run. To load every night with nobody watching, it needs a parameter, a pipeline that runs it on a schedule and reads its answer, and a little upkeep for the tables it writes.

#A parameter cell

Put the table's name in its own cell at the top of the notebook:

table_name = "orders"

Open the cell's menu and choose Toggle parameter cell. A pipeline can now pass another value in, and the rest of the notebook uses table_name instead of the word orders, so one notebook can load several tables with their own watermark rows.

#The pipeline

In the workspace, create a Data pipeline and add a Notebook activity named Load orders:

• Settings: pick the workspace and the notebook.
• Base parameters: add table_name, type String, value orders.

Run it once from the pipeline. The run's output shows the notebook's exit value, the number of rows it merged.

#Read its answer

The number the notebook exits with is in the activity's output, for the next step to use:

@activity('Load orders').output.result.exitValue

Add an If condition after it: when the value is 0, nothing was new. That is fine on a Sunday and worth a look on a Tuesday, so the false branch can send a message with an Office 365 Outlook or Teams activity.

#The schedule

Press Schedule in the pipeline's menu, turn it on, and choose a time after the bronze load has finished, for example every day at 02:00 in your time zone. Every run is listed in the pipeline's Run history and in the Monitor hub.

#Upkeep

A table merged into every night collects many small files. Once a week, a short notebook or a second scheduled step keeps it fast:

spark.sql("OPTIMIZE silver_orders")
spark.sql("VACUUM silver_orders RETAIN 168 HOURS")

OPTIMIZE packs small files into bigger ones. VACUUM removes the files older versions no longer need after seven days, so time travel still works for a week. The lakehouse also offers both under Maintenance in a table's menu.

#What you built

A watermark table, a notebook that merges only the rows since the last good run and moves the watermark on only after it succeeded, and a pipeline that runs it every night and says when nothing came in.