Build a streaming pipeline: Kafka to bronze, silver and an hourly gold table, checked

This part needs everything from parts 1 to 3: Kafka running in Docker, and stream_session.py, make_orders.py, read_orders.py, latest_orders.py and hourly_sales.py in 1 folder. It starts again from a new topic, so nothing from earlier runs is needed.

Build it: the shop's streaming pipeline, checked

The pipeline is 3 streams, each loading what is new and stopping:

Add 1 more batch to BATCHES in make_orders.py, an order an hour after the others, which moves the watermark on:

    "5": [
        (1009, "C03", "placed", 30.00, "2026-10-01T13:20:00Z"),
    ],

Save this as run_streams.sh, which runs the streams in order, as a job would after new events arrive:

#!/bin/sh
# Run the 3 streams in order: Kafka to bronze, bronze to silver, bronze to gold. Each loads what is new and stops.
set -e
python read_orders.py
python latest_orders.py
python hourly_sales.py

A check for every table

A stream's tables can be worked out again from bronze with an ordinary batch query: the newest event of each order, and each hour's count. Save this as check_streams.py:

"""Check the 3 tables the streams fill against the raw events in bronze_orders. Exits 1 on a problem."""
import sys
from pyspark.sql import functions as F
from stream_session import stream_spark

spark = stream_spark("check_streams")
bronze = spark.table("bronze_orders")
problems = 0


def orders(n):
    return f"{n} order" if n == 1 else f"{n} orders"


# 1. Bronze keeps every event Kafka holds, each offset once.
events = bronze.count()
if events == bronze.select("partition", "offset").distinct().count():
    print(f"ok      bronze_orders has {events} events, each offset once")
else:
    print("PROBLEM bronze_orders has an offset more than once: a stream read it twice")
    problems += 1
# An event sent twice gets a new offset, so bronze keeps both. Silver and gold must count it once.
distinct = bronze.dropDuplicates(["order_id", "status", "event_time"])
repeats = events - distinct.count()
if repeats:
    print(f"note    {repeats} of them sent twice by the shop's system")

# 2. Silver: orders holds the newest event of each order, worked out again from bronze.
newest = (
    distinct.groupBy("order_id")
    .agg(F.max_by(F.struct("customer_id", "status", "amount", "event_time"), "event_time").alias("event"))
    .select("order_id", "event.*")
)
silver = spark.table("orders")
if newest.exceptAll(silver).isEmpty() and silver.exceptAll(newest).isEmpty():
    print(f"ok      orders has the newest state of all {silver.count()} orders")
else:
    print("PROBLEM orders differs from the newest events in bronze")
    problems += 1

# 3. Gold: each hour written matches a recount of the different events, except those later than the watermark.
recount = (
    distinct.where("status = 'placed'")
    .groupBy(F.date_trunc("hour", "event_time").alias("hour"))
    .agg(F.count("*").alias("all_orders"), F.sum("amount").alias("all_revenue"))
)
for row in recount.join(spark.table("hourly_sales"), "hour", "left").orderBy("hour").collect():
    hour, counted, known = row["hour"].strftime("%H:%M"), row["orders"], row["all_orders"]
    if counted is None:
        print(f"waiting {hour} is not written yet: the watermark has not passed its end")
    elif counted == known:
        print(f"ok      {hour} {orders(counted)}, {row['revenue']}")
    elif counted > known:
        print(f"PROBLEM {hour} counted {orders(counted)}, {row['revenue']}, but bronze has {orders(known)}: an event counted twice")
        problems += 1
    else:
        print(f"late    {hour} counted {orders(counted)}, {row['revenue']}; bronze has {orders(known)}, {row['all_revenue']}: {known - counted} arrived after the hour was written")

sys.exit(1 if problems else 0)

It tells 3 kinds of difference apart. An hour that counts fewer orders than bronze is late: the watermark dropped an event, as designed, and a nightly recount corrects it. An hour that counts more is a problem: something was counted twice. A missing hour is waiting for the watermark.

Run it

Start from a new topic and nothing on disk, then send the 5 batches, running the streams after each. Batch 3 goes twice, as the shop's system would send it again after a timeout. Kafka's producer stops its own retries from writing a message twice, but a second send by the application is a new message:

docker exec --workdir /opt/kafka/bin kafka ./kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic orders
docker exec --workdir /opt/kafka/bin kafka ./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3 --replication-factor 1
rm -rf checkpoints spark-warehouse metastore_db
for batch in 1 2 3 3 4 5; do python make_orders.py $batch && ./run_streams.sh; done
python check_streams.py

ok      bronze_orders has 11 events, each offset once
note    1 of them sent twice by the shop's system
ok      orders has the newest state of all 9 orders
late    09:00 counted 3 orders, 68.50; bronze has 4 orders, 90.50: 1 arrived after the hour was written
ok      10:00 2 orders, 55.00
PROBLEM 11:00 counted 2 orders, 36.50, but bronze has 1 order: an event counted twice
waiting 12:00 is not written yet: the watermark has not passed its end
waiting 13:00 is not written yet: the watermark has not passed its end

Bronze is right to keep both copies of order 1005: it holds what Kafka holds, at 2 offsets. Silver is right too, because its MERGE found 1005's second event no newer than the first and changed nothing. The hourly count added it again. The fix is part 2's dropDuplicatesWithinWatermark, in hourly_sales.py after the watermark:

    .withWatermark("event_time", "1 hour")
    # The shop's system can send an event twice. Count it once, remembering each event as long as the watermark allows.
    .dropDuplicatesWithinWatermark(["order_id", "status"])

The query now has 2 parts that keep state, the duplicates and the hours, and either can drop a late event; part 2's progress loop adds up what each drops, so it still reports order 1006 as too late. Run the same commands again:

ok      bronze_orders has 11 events, each offset once
note    1 of them sent twice by the shop's system
ok      orders has the newest state of all 9 orders
late    09:00 counted 3 orders, 68.50; bronze has 4 orders, 90.50: 1 arrived after the hour was written
ok      10:00 2 orders, 55.00
ok      11:00 1 order, 18.25
waiting 12:00 is not written yet: the watermark has not passed its end
waiting 13:00 is not written yet: the watermark has not passed its end

The check exits 0. The 09:00 hour is still 22.00 short, order 1006, and will stay so in hourly_sales; part 2's recount.py is the batch that corrects it. Batch 5 moved the watermark to 12:20, which closed the 11:00 hour, and an order after 14:00 would close 12:00.

On Databricks

The 3 streams become 1 job, scheduled every 15 minutes or every hour, each stream a task with availableNow and the one before it as its dependency, and check_streams.py as the last task, so a problem fails the run where someone sees it, as module 8 did for serving. Module 10 builds that job. Name the tables in Unity Catalog and put each checkpoint in a volume, as part 4 shows; read Kafka with a service credential, or let a streaming table read it with read_kafka. check_streams.py reads only tables, so it runs on serverless compute with the notebook's spark in place of stream_spark and the tables' Unity Catalog names.

Which streaming service to put in front is its own choice, with its own costs:

ComparisonKafka vs Kinesis vs Pub/SubKafka, Kinesis Data Streams and Pub/Sub all carry events from the services that write them to the services that read them. Kafka is an open-source log that you…