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

A streaming pipeline is right when every table it fills agrees with the raw events, apart from what the watermark says it drops. The shop's pipeline reads Kafka into bronze_orders, keeps each order's newest state in orders with a MERGE, and counts each hour in hourly_sales. check_streams.py works out silver and gold again from bronze and compares them. Replayed with 1 batch sent twice, it caught the hourly table counting an order twice, which dropDuplicatesWithinWatermark fixed; the silver MERGE was already safe.

- Page: https://tecyfy.com/data-engineering/course/streaming/build-it
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 9: Streaming, Build it: orders from Kafka to an hourly table
- Parts of this module: Kafka into a Delta table (https://tecyfy.com/data-engineering/course/streaming); Event time, windows and watermarks (https://tecyfy.com/data-engineering/course/streaming/event-time); Exactly-once and foreachBatch (https://tecyfy.com/data-engineering/course/streaming/exactly-once); Streaming on Databricks (https://tecyfy.com/data-engineering/course/streaming/on-databricks); Build it: orders from Kafka to an hourly table (https://tecyfy.com/data-engineering/course/streaming/build-it)

At the end of this part you can:

- Build a streaming pipeline from Kafka to bronze, silver and gold, each stream loading what is new and stopping
- Check every table a stream fills against the raw events, and tell an event dropped by the watermark from one counted twice
- Replay the whole pipeline from a new topic, and say how its steps run as a job on Databricks

This part needs everything from parts [1](/data-engineering/course/streaming) to [3](/data-engineering/course/streaming/exactly-once): 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:

| Stream | Reads | Writes | Why it is safe to run again |
| --- | --- | --- | --- |
| `read_orders.py` | the Kafka topic `orders` | `bronze_orders`, every event | Delta's sink and the checkpoint's offsets |
| `latest_orders.py` | `bronze_orders` | `orders`, the newest state of each order | a `MERGE` on the key, newer events only |
| `hourly_sales.py` | `bronze_orders` | `hourly_sales`, orders and revenue per hour | Delta's sink, and the watermark closes each hour once |

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

```python
    "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:

```bash
#!/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`:

```python
"""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:

```bash
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
```

```text
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:

```python
    .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:

```text
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](/data-engineering/course/serving-data/build-it) did for serving. Module 10 builds that job. Name the tables in Unity Catalog and put each checkpoint in a volume, as [part 4](/data-engineering/course/streaming/on-databricks) 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:

- Comparison: [Kafka vs Kinesis vs Pub/Sub](https://tecyfy.com/data-engineering/kafka-vs-kinesis-vs-pub-sub). Kafka, 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…

## Check yourself

1. With batch 3 sent twice, orders was right but hourly_sales counted order 1005 twice. Why did 1 table handle the repeat and not the other?

   A. orders was rebuilt from scratch
   B. Kafka removed the repeat before silver read it
   C. The MERGE matched 1005's second event on its key, found it no newer, and changed nothing; the hourly count adds every event it reads
   D. hourly_sales reads a different topic

2. check_streams.py reports "late 09:00 counted 3 orders, 68.50; bronze has 4 orders, 90.50" and still exits 0. Why is that not a problem?

   A. The check is broken
   B. Order 1006 arrived after the watermark had closed the 09:00 hour, which the stream is designed to drop; bronze keeps it, and a nightly recount corrects the hour
   C. 1006 was a duplicate
   D. hourly_sales has not run yet

### Answers

1. C. A MERGE on the key with a newer-event condition is idempotent, so a repeat changes nothing. A count is not: each event read adds 1. dropDuplicatesWithinWatermark(["order_id", "status"]) makes the count skip an event it has seen, for as long as the watermark keeps it.
2. B. The watermark trades late events for an hour that is final. A late event is what the design expects; an event counted twice is not, which is why one is a note and the other a problem. Part 2's recount.py shows how a batch puts the late order back.

## Questions

### How do I test a Spark Structured Streaming pipeline?

Send known events, run each stream with trigger(availableNow=True) so it loads what is there and stops, then compare each table it fills with a batch query over the raw events, such as the newest event per key for an upserted table and a recount per window for an aggregate. Replay from a new topic and new checkpoints, and include the cases that break streams: an event sent twice, an event later than the watermark and a job that fails after writing.

### Does Kafka deliver each message once?

Each event is stored at 1 offset in its partition, and a stream with a checkpoint reads each offset once. Kafka's producer is idempotent by default, so its own retries do not write a message twice. But an application that sends an event again, after a timeout or a restart, stores it twice, at 2 offsets, and the stream reads both. Make what follows safe for it: a MERGE on the key for current state, and dropDuplicatesWithinWatermark on the event's key before a count.

## Sources

- [Structured Streaming Programming Guide: APIs on DataFrames and Datasets](https://spark.apache.org/docs/latest/streaming/apis-on-dataframes-and-datasets.html)
- [Structured Streaming Programming Guide, getting started](https://spark.apache.org/docs/latest/streaming/getting-started.html)
- [Kafka producer configs](https://kafka.apache.org/43/configuration/producer-configs/)
- [Production considerations for Structured Streaming](https://docs.databricks.com/aws/en/structured-streaming/production)
- [Configure Structured Streaming trigger intervals](https://docs.databricks.com/aws/en/structured-streaming/triggers)