Payments data every hour instead of once a day

The situation

Payments data reached Axi’s lakehouse through a daily job that reloaded the full history on every run. It took 95 minutes, and because it ran once a day, a payment could wait up to 24 hours before finance and risk saw it.

As history grew, so would the run time. The fix wasn’t a bigger cluster. It was to stop reprocessing data that hadn’t changed.

What changed

1. Read only what changed. Each run picks up rows updated since the last successful run, using a watermark on the last-updated timestamp.
2. Merge by business key. Changes merge into the Delta table by payment ID, so a rerun updates rows instead of duplicating them.
3. Run every hour. With a run that small, an hourly schedule costs little and cuts the delay to under an hour.
4. Ship it as code. The job, its schedule and its compute live in a Databricks Asset Bundle, deployed through Azure DevOps CI like any other change.

Key decisions

• Watermark on updates, not inserts

  Payments change state after they’re created. Tracking the last-updated time means status changes reach reporting too, not only new payments.

• Safe to rerun

  Merging on payment ID makes every run idempotent, so a failed hour can simply run again.

• Hourly batch, not streaming

  Finance and risk needed numbers within the hour, not by the second. An hourly batch met that with far less to operate than a streaming job.

• Job settings in the repository

  With Asset Bundles, schedule and cluster changes go through pull requests instead of being edited by hand in the workspace.

What it looks like in code

from delta.tables import DeltaTable
from pyspark.sql import functions as F
from pyspark.sql.window import Window

SOURCE = "finance.bronze.payments_raw"
TARGET = "finance.silver.payments"
OVERLAP_MINUTES = 15  # re-read a little to catch late updates

target = DeltaTable.forName(spark, TARGET)
watermark = spark.table(TARGET).agg(F.max("updated_at")).first()[0]

changes = spark.table(SOURCE)
if watermark is not None:
    since = F.lit(watermark) - F.expr(f"INTERVAL {OVERLAP_MINUTES} MINUTES")
    changes = changes.where(F.col("updated_at") > since)

# Keep only the newest version of each payment in this batch
latest = (
    changes
    .withColumn("rn", F.row_number().over(
        Window.partitionBy("payment_id").orderBy(F.col("updated_at").desc())))
    .where("rn = 1")
    .drop("rn")
)

(
    target.alias("t")
    .merge(latest.alias("s"), "t.payment_id = s.payment_id")
    .whenMatchedUpdateAll(condition="s.updated_at > t.updated_at")
    .whenNotMatchedInsertAll()
    .execute()
)

resources:
  jobs:
    payments_hourly:
      name: payments-hourly
      schedule:
        quartz_cron_expression: "0 5 * * * ?"   # five past every hour
        timezone_id: UTC
      job_clusters:
        - job_cluster_key: small
          new_cluster:
            spark_version: 15.4.x-scala2.12
            node_type_id: Standard_D4ds_v5
            num_workers: 1
      tasks:
        - task_key: merge_payments
          job_cluster_key: small
          spark_python_task:
            python_file: ./src/merge_payments.py

Illustrative sample written for this page, with made-up names. It shows the pattern, not client code.

If you’re making the same change

• Check the source can tell you what changed. Incremental loads need a reliable updated-at column or change feed. Confirm it exists before designing around it.
• Re-read a small overlap. Late updates can land just behind the watermark. Re-reading the last few minutes on each run is cheap insurance.
• Keep a full reload for recovery. A backfill job that rebuilds history is still worth having. It just shouldn’t be the daily path.