Streaming: Picture, Rhythm, Example
Concepts covered: Stream Processing
A streaming pipeline also has 3 parts beyond the data: a queue that holds records in arrival order, a consumer process that reads them, and a sink that receives the results. The consumer plays the roles of both the scheduler and the job from the batch picture. It is started once and expected to stay up for months. There is no trigger time to look up, because the arrival of a record is the trigger.
- signup app (source, web application)
- signups queue (queue, message queue)
- stream counter (transform, stream processor)
- live_counts (storage, serving table)
- live dashboard (consumer, ops screen)
- signup app -> signups queue
- signups queue -> stream counter
- stream counter -> live_counts
- live_counts -> live dashboard
One event, source to sink
Follow signup s2, a user in Germany who presses the signup button at 07:40:00 on April 24.
| Clock | Where s2 is | What happens |
|---|---|---|
| 07:40:00.000 | Signup app | The app saves the user and publishes `{id: s2, country: DE}` to the queue |
| 07:40:00.040 | Queue | The queue appends the record after `s1` and offers it to consumers |
| 07:40:00.120 | Stream counter | The consumer, blocked waiting for input, receives `s2` |
| 07:40:00.125 | Stream counter | It reads the current DE count from the sink, 0, and computes 1 |
| 07:40:02.000 | `live_counts` | It writes the row DE, 1 and tells the queue that `s2` is done |
| 07:40:02.300 | Live dashboard | The next screen refresh reads DE, 1 |
Two seconds after someone in Germany pressed a button, a screen in the office shows the new count, and nothing was scheduled to make that happen. The consumer was waiting at 07:39, it processed s2 at 07:40, and it returned to waiting at 07:40:02 for s3, which arrives at 11:05. For 3 hours and 25 minutes the process is alive, allocated, billed, and doing nothing except waiting.
for event in queue.subscribe("signups"):
country = event["country"]
current = live_counts.get(country, default=0)
live_counts.upsert(country, current + 1)
queue.mark_done(event["id"])That for loop is the entire transform. queue.subscribe blocks until a record is available, so the loop body runs once per signup and never on a timer. live_counts.get fetches the current count for one country. upsert inserts that country's row or replaces it, 1 row rather than the whole table. mark_done tells the queue the consumer has finished s2, so a consumer that crashes and restarts resumes at s3 instead of rereading the day from s1.
The queue is the shock absorber
The queue sits between the app and the consumer so that neither waits for the other. The app publishes a record in 40 milliseconds and goes back to serving the user. If the consumer is taken down for an upgrade at 15:00, signup s4 at 15:30 lands in the queue and sits there. When the consumer returns at 15:45 it reads s4 and carries on. The queue turned a 45 minute outage into a 15 minute delay for 1 record, and the app never noticed. Without the queue, the app would have had to wait for the counter or drop the record.
- Batch job down
- The next read is hours away, so nobody sees it
- A rerun reads the same closed window
- Same output, finished later
- Consumer down
- New records wait in the queue
- Readers see a frozen, stale table
- Restart resumes at the first unfinished record
Signups feed both pipelines from the animation. At 15:00 on April 24 the stream consumer is stopped for an upgrade and restarted at 15:45. The nightly job runs as usual at 02:00 on April 25. At 02:10 on April 25, what does each table show for IN?
- live_counts has no IN row and daily_counts shows IN 1, because the stream missed s4 while it was down.
- Both show IN 1: the queue held s4 until the consumer returned at 15:45, and the batch read it at 02:00.
- live_counts shows IN 1 and daily_counts has no IN row, because s4 arrived while the consumer was down.
- Both show IN 2, because the restarted consumer reprocessed s4 and the batch counted the duplicate.
On a day with 6 signups, the consumer works for about 12 seconds and waits for the other 23 hours, 59 minutes and 48 seconds. The rhythm is streaming regardless of volume.
At 11:05 signup s3, from the US, lands in the queue. At 11:05:05 on April 24, what do the 2 tables show for US?
- live_counts shows US 2 and daily_counts shows US 1, because the batch wrote the overnight count at 02:08.
- live_counts shows US 1 and daily_counts shows US 1, because both tables update within seconds of a record.
- live_counts shows US 1 and daily_counts is empty, because the stream has not caught up with s3 yet.
- live_counts shows US 2 and daily_counts has no rows, because its first run happens at 02:00 on April 25.
About this lesson
This section is part of the Batch vs Streaming: Beginner lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.