How Do You Restart a Stream?

A checkpoint is a snapshot of everything the job would need to continue as if it had never stopped. It holds 3 things written together: the offset in each source partition the job has consumed up to, the state of every operator at that offset (open window totals, the set of seen ids, the join buffer), and the position the sink has acknowledged. The 3 are written as 1 unit to durable storage, so that a restart never pairs state from 1 moment with offsets from another.

A crash, second by second

The job reads 2,000 events a second across 12 partitions, aggregates into 5 minute windows, and checkpoints every 30 seconds. The table follows a crash from the last good snapshot to full recovery.

The events between offsets 1,060,000 and 1,090,000 are 15 seconds of traffic at 2,000 a second. The job consumed them and emitted some before the crash, and the restarted job consumes them again. Their totals are added to state restored from checkpoint 42, which does not include them, so the aggregate stays correct. Nothing is lost because the log still holds them and the checkpoint recorded where to resume. What moves is the latency: a consumer watching the sink sees no update for about 55 seconds, from the crash to catch-up, and then a burst.

• 30,000
• 60,000

What the sink sees twice

Correct state after restart is 1 half of the problem. The sink received emissions for the 10:00 window before the crash, perhaps a running total of 88,000 at 10:00:44, and after replay it receives emissions covering the same events again. An append-only sink now holds 2 rows for the same window. 3 designs absorb this. An upsert sink keyed by product and window start overwrites the earlier row, so the latest emission wins. A transactional sink commits output in the same transaction as the offsets, so the writes after checkpoint 42 were never committed. A dedupe view on the read side keeps the latest emission per key, as below. The guarantee each design gives, and what each costs, is the subject of the delivery semantics lesson.

SELECT
  product_id,
  window_start,
  order_count
FROM (
  SELECT
    product_id,
    window_start,
    order_count,
    row_number() OVER (
      PARTITION BY product_id, window_start
      ORDER BY emitted_at DESC
    ) AS rn
  FROM stream_sink
) AS ranked
WHERE rn = 1

The emitted_at column makes the view work: the replayed emission for a window is written after the orphaned pre-crash write, so the latest row wins and the orphan stays in the table without being read. The cost is a window function on every query, which is why an upsert sink is the usual choice when the sink supports it.

Choosing the checkpoint interval

The interval sets the size of the replay and the write load on the checkpoint store, and the 2 pull in opposite directions. At 30 seconds, the store receives a snapshot of the state twice a minute. With 2 GB of state that is 4 GB a minute of writes for a full snapshot, which is why engines write incremental checkpoints holding only the keys changed since the last one. At 5 seconds, replay shrinks to 10,000 events, but every snapshot still pauses operators to write offsets, and that fixed cost runs 6 times as often. The rule to give in an interview: pick an interval that is a small fraction of the latency SLA, often a tenth, bounded below by what the store can absorb, and measure recovery time with a deliberate kill in staging.

About this lesson

This section is part of the Batch or Stream?: Intermediate 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.