Batch & stream processing

One computation, two kinds of input

Batch and stream processing look like two technologies, and the useful way to see them is as one idea applied to two kinds of input. Tyler Akidau's framing, which most of the field has adopted, splits data by finiteness, not by engine: a bounded data set has an end (yesterday's click log, a table snapshot) and an unbounded one does not (clicks as they happen). A batch job is what you run when the input is bounded; a stream job is what you run when it is not. Run a stream engine on a bounded input and you get a batch job. The reverse is where people go wrong: running a nightly job every minute is not streaming, it is a batch system with a short fuse.

The practical rule: a bounded input lets the engine use perfect knowledge. It can sort by timestamp and wait for the end of the data, so it never has to guess when a window is complete. An unbounded input forces the guess, and the rest of this chapter is about making it well.

From MapReduce to Spark: what actually changed

MapReduce (Dean and Ghemawat, OSDI 2004) is a programming model with two user-supplied functions: map turns input key/value pairs into intermediate key/value pairs, and reduce merges all the values that share an intermediate key. In between, the runtime does the shuffle: it partitions the map output by key, moves it across the network and groups it, so that every value for one key lands at the same reducer. The programmer writes two functions; the runtime handles partitioning, scheduling, machine failure and communication. The paper reports that a typical computation processes many terabytes on thousands of machines, and that over a thousand jobs a day ran on Google's clusters.

The model's limit is the shape of real pipelines. One MapReduce job is one map-shuffle-reduce; a useful analysis is a chain of them, and in the classic Hadoop arrangement each job reads its input from a distributed file system and writes its output back to one. Iterative algorithms (graph ranking, machine learning) pay that disk round trip on every pass. Spark, from the Berkeley paper on Resilient Distributed Datasets (NSDI 2012), attacked exactly that: keep the working data in memory across steps, which the authors say can improve performance by an order of magnitude for iterative and interactive work. The trick is how it stays fault tolerant without replicating memory. An RDD is built by coarse-grained transformations (map, filter, join) applied to a base data set, and Spark records the chain of transformations. If a partition is lost, Spark recomputes it from the transformations that produced it, the lineage, instead of restoring a copy.

Log-based streaming: the log is the source of truth

Stream processing started as message-passing between operators. The model that won puts a replicated log in the middle (see the message queues chapter for queues against logs). Events are appended to a Kafka topic; a stream engine such as Flink or Kafka Streams reads it with a consumer position that is just an offset, computes, and writes results to another topic or a store. Three properties of the log carry the whole design:

• Replay. The input is retained, so a job that crashes, or ships a bug, can be pointed back at an older offset. Reprocessing is a first-class operation, not a disaster procedure.
• Many independent readers. The fraud job, the dashboard job and the warehouse loader each keep their own offsets and cannot slow one another.
• A position that is one number per partition. A consumer's position in a partition is a single integer, the offset of the next message to read, which makes checkpointing it cheap. That is what lets a stream engine snapshot "how far I have read" together with "what I have computed".

The idea underneath is stream-table duality: a stream is the changelog of a table (replay it from the start and you rebuild the table), and a table is a snapshot of the latest value per key that can be turned back into a stream. Log compaction, covered in the queues chapter, is that duality implemented in storage. It matters later, because it is why a compacted topic can serve as the rebuild source for a derived view.

Event time against processing time

Every record has two clocks. Event time is when the thing happened on the device that produced it; processing time is when your system observes it. Flink's docs describe processing time as simplest and lowest latency but nondeterministic, because it depends on the speed at which records arrive; in event time, progress depends on the data, not on any wall clock. The gap between the two, the skew, is never zero and never constant: a phone in a tunnel delivers a click two minutes late, a Kafka consumer that fell behind delivers everything late, and a mobile client batches events and uploads them every few minutes.

If you aggregate by processing time, "clicks in the 12:00 minute" means "clicks my pipeline happened to see during 12:00", which changes when the pipeline is slow, restarts or replays. Billing, ranking and anything you will be audited on must use event time. The cost is that event-time answers are never obviously complete, which is the problem the next section solves.

Watermarks: a heuristic for "nothing older is coming"

A window over event time has to close some time. The engine cannot know that no late record is on its way, so it estimates. A watermark is a marker that flows through the stream with a timestamp t and the claim "event time has reached t; elements at or below t should no longer arrive". Flink generates them at or near the sources; a window fires once the watermark passes its end. The common strategy is bounded out-of-orderness: the watermark is the largest event timestamp seen so far, minus a fixed bound (Flink's own example emits currentMaxTimestamp - maxOutOfOrderness - 1 milliseconds). Spark Structured Streaming defines it the same way: the maximum event time seen, minus a threshold you set, such as 10 minutes.

Three consequences are worth saying out loud, because each is a trade-off:

• The bound is a dial between latency and completeness. A 10-second bound delays every window result by about 10 seconds past the window end, and still loses anything later than that. A 10-minute bound loses far less and delays every result by 10 minutes. There is no correct value, only the lateness distribution of your data and what a wrong answer costs. Akidau calls watermarks "a reasonably accurate heuristic estimate", not a proof.
• The minimum rules, so one slow input stalls everything. An operator with several inputs, or one after a keyBy, takes the minimum of its inputs' watermarks (Spark's default policy for multiple streams is also the minimum). An input that goes quiet, such as an idle Kafka partition, holds the minimum back and no window anywhere fires. Flink's answer is withIdleness, which marks an input idle after a timeout so it stops holding the watermark back.
• Spark's guarantee is one-sided. Data less than the threshold behind the latest processed event time is always aggregated; data later than that may or may not be, and the more delayed, the less likely. Do not design as if late means dropped.

What happens to a late record is a policy, not an accident. Flink's default allowed lateness is zero: once the watermark is past the window end, the late element is dropped. With allowed lateness greater than zero, a late element that still falls inside the window's end plus that lateness is added and the window fires again with an updated result, so downstream must tolerate updates. Past that, you can route the element to a side output instead of discarding it silently, and let a batch job reconcile it later.

Windows: three shapes of "per minute"

Do the arithmetic before choosing. A one-hour window that slides every minute puts each event in 60 windows. At a steady 11,574 events a second that is 11,574 × 60 = 694,440 window assignments a second, and each key holds 60 open windows of state. Engines reduce the work by sharing partial results between windows, but the state and the emit rate still scale with length ÷ slide. A session example, gap 30 minutes: clicks at 10:00, 10:10 and 10:50 give two sessions. The first spans 10:00 to 10:40 (the last event plus the gap), and the 10:50 click is 40 minutes after 10:10, so it starts a second, spanning 10:50 to 11:20. If a late click stamped 10:30 then arrives, its own window [10:30, 11:00) overlaps both, and the sessions merge into one that spans 10:00 to 11:20. That merge is why session windows need a merging trigger and are the shape that late data disturbs most.

State and checkpoints: how a stream job survives a crash

A window count, a deduplication set, a join buffer: all are state, held in the operators, and a job that cannot recover it is useless for anything beyond a stateless filter. Flink's approach, from the Carbone et al. paper on asynchronous barrier snapshotting, is this. The checkpoint coordinator tells the sources to record their offsets and insert numbered barriers into their streams. A barrier flows with the data; when an operator sees barrier n it snapshots its state, and an operator with two inputs aligns them, so checkpoint n contains the effect of every event before barrier n and none after it. The result is a consistent cut: the source offsets and the operator state that correspond to exactly those offsets. Processing carries on while snapshots are written out asynchronously.

Recovery follows directly. Restore every operator's state from the last completed checkpoint, rewind each source to the checkpointed offset, and continue. Records between the checkpoint and the crash are processed again, but against state that has forgotten them, so state is not double counted. State size decides the backend: Flink's heap-based backend keeps state in JVM memory (fast, limited by memory, GC pressure), while the RocksDB backend keeps it on local disk so it can exceed memory and supports incremental snapshots, at a speed cost the docs put at roughly ten times slower than heap.

Dry run: crash and catch-up, with planning numbers

Assumptions, all planning figures: the job reads 11,574 events a second on average, checkpoints every 30 seconds, crashes 20 seconds after the last checkpoint, takes 60 seconds to restart, and can process at twice the arrival rate once it is running again.

Two lessons. Catch-up time is (downtime + replay window) ÷ (capacity ÷ arrival − 1): with no headroom (capacity equal to arrival) a job never catches up, so size stream jobs with spare capacity. And frequent checkpoints shrink the replay window but cost I/O and, with an exactly-once Kafka sink, they are also how often output becomes visible.

Delivery guarantees, and what "exactly-once" means here

The queues chapter showed why exactly-once delivery is impossible and exactly-once effect is built from at-least-once plus idempotence or transactions; read that first, and here is how a stream engine applies it. Flink's documentation is blunt about the term: exactly-once does not mean each event is processed once, it means every event affects the Flink-managed state exactly once, because recovery replays source data. End to end, you need three things:

• A replayable source (a Kafka offset you can rewind to).
• State restored from a consistent checkpoint, so the replay does not double count.
• A transactional or idempotent sink, so output written for records that will be replayed does not appear twice.

For Kafka output the engine uses Kafka's own machinery: the idempotent producer deduplicates retries with sequence numbers, and transactions let a producer write across partitions atomically (both are in the queues chapter). Flink's Kafka sink in EXACTLY_ONCE mode writes records inside a Kafka transaction that commits when the checkpoint completes, which has three operational consequences from the connector docs: downstream consumers must use isolation.level=read_committed or they will see records that may be aborted; records are not visible until a checkpoint completes, so checkpoint interval sets output latency; and the docs strongly recommend a producer transaction timeout larger than the maximum checkpoint duration plus the maximum restart time, because otherwise Kafka can expire an uncommitted transaction and data may be lost.

Lambda against Kappa

The Lambda architecture answers "streaming is fast but approximate, batch is slow but correct" by running both over the same immutable input and merging the results at query time. Jay Kreps' 2014 critique names the cost: the transformation logic must be implemented twice, in two complex distributed systems, and kept equal. His words are that he does not think the problem is fixable, and that abstraction layers compiling one program to both engines helped little. Akidau went further: well-designed streaming systems are a strict superset of batch, and Lambda belongs to the history of big data.

The alternative is the one Kreps proposed in the same post and half-jokingly suggested calling the Kappa Architecture ("it may be too simple of an idea to merit a Greek letter"): one pipeline and a retained log. To change the logic you start a second instance of the same job from the beginning of the log, writing to a new output table; when it catches up you switch readers to the new table, then stop the old job and delete the old table. The code is written once, and reprocessing is something you do every time the code changes.

Notice what the table does not say: batch is not obsolete. Batch is the right tool when the input is bounded and efficiency matters, and a nightly reconciliation that recounts the day exactly is how you catch what the stream dropped. The modern pattern is one logic, two execution modes (Flink's batch mode, Spark's unified APIs), not two programs.

Change data capture and derived data

Every system has a primary store, and then a growing set of derived data: the search index, the cache, the analytics warehouse, the recommendation features. The naive way to keep them current is for the application to write to each one (a dual write). It fails the way the queues chapter's outbox section shows: the writes are not atomic across systems, so a crash between them leaves the index and the database disagreeing, and two concurrent writers can apply their updates in opposite orders to the two stores, so they disagree permanently.

Change data capture fixes it by making the database's own change log the single source. Log-based tools such as Debezium read the database's change stream (the MySQL binlog, the PostgreSQL logical replication stream) and publish the events, generally to one Kafka topic per table. Every derived store then consumes that topic and applies the changes in order, so each is a deterministic function of one log. Three design points follow:

• Expect duplicates. Debezium's FAQ states that after a crash consumers may see events more than once. Apply changes idempotently: upsert by primary key, and carry the source position so a stale replay cannot overwrite newer data.
• Bootstrap, then tail. A new derived store needs the existing rows, not just future changes. Either snapshot the table and then switch to the log at the matching position, or, for a table that is read as a changelog, replay a compacted topic, which keeps the latest record per key.
• Deletes are events. A delete must reach every derived store; in a compacted topic that is a record with a key and a null value (a tombstone), and a consumer that stays away longer than the delete-retention period can miss it.

The payoff is the reprocessing story again: to add a new index, or to fix a bug in how one is built, you replay the log into a fresh instance and cut over. That is why teams that adopt a log tend to adopt CDC, and why the outbox from the queues chapter and CDC are the same idea at two scales: a transactionally written change record, then a relay that is allowed to repeat itself.

Worked example: ad-click aggregation

The prompt (it is a chapter of Alex Xu's System Design Interview, Volume 2; the framing here is mine): advertisers want to know how many times each ad was clicked, per minute, within seconds, and they will be billed from the numbers. That sentence sets the constraints. The counts must be correct (money), nearly real-time (advertisers watch campaigns), tolerant of late and duplicate events (phones, retries), and recoverable (a crashed job cannot lose or double count a click). The figures below are this chapter's assumptions, not quotations from any source, so change them if the interviewer does.

Capacity estimation

• Volume: 1 billion clicks a day ÷ 86,400 s = 11,574 clicks a second on average. Allow a 5× peak: about 57,870 a second.
• Raw storage: 100 bytes per click event × 1 billion = 100 GB a day, or 36.5 TB a year before replication. Keep it: raw events are what you replay and reconcile from.
• Ingest bandwidth at peak: 57,870 × 100 B = 5.8 MB/s. Under the 10 MB/s per-partition planning figure from the queues chapter, so volume alone does not force many partitions; parallelism for the aggregation does.
• Aggregated output: assume 50,000 distinct ads are clicked in an average minute. One row per ad per minute is 50,000 × 1,440 = 72 million rows a day; at 40 bytes a row that is 2.88 GB a day, about 35 times smaller than the raw data. (At 694,444 clicks a minute against 50,000 rows, each row summarises about 14 clicks.)
• Partitions: if one aggregation task handles about 10,000 events a second, peak needs 57,870 ÷ 10,000 ≈ 6 tasks. Choose 12 partitions, which gives about 2× headroom and room to grow, and partition the topic by click id, not by ad id (the reason is in the hot-key bullet below).

The decisions, in the order an interviewer pushes

• Where does deduplication happen? Clients retry and some are malicious. Stamp each click with a unique click id at the click service and partition the topic by it, so every duplicate of a click lands in the same partition and one task can keep a deduplication set in its own state (bounded by the lateness window plus a margin, or it grows without limit). Combine with the exactly-once chain: replayable source, checkpointed state, idempotent sink.
• What is the hot-key answer? A viral ad can take, say, 20% of all clicks: 0.2 × 57,870 = 11,574 clicks a second, more than the 10,000 a second one task handles. If the topic were partitioned by ad id, all of it would land in one partition, and no downstream trick helps a task that must still read every event. Because the topic is partitioned by click id, the hot ad's clicks arrive spread across all 12 partitions. Then use two-stage aggregation: each task pre-aggregates its own slice per (ad, minute), and a second stage, keyed by ad id, merges the partial counts. That is the combiner idea from MapReduce, and counts are additive, so partials merge exactly. The second stage now receives at most 12 partials per ad per minute instead of every click. When a late event updates a task's partial, that task re-emits its cumulative partial and the second stage replaces its previous value from that task before summing.
• Why event time, and how much lateness? Billing must reflect when the click happened. Pick the watermark bound from the measured lateness distribution (the 99th percentile of skew, say), keep a longer allowed lateness for updates, and send what is left to the side output.
• Why also a batch path? Because the stream's answer is an estimate of "done" by construction. The batch job recounts from the raw archive for finished days with perfect knowledge of arrival (a bounded input needs no watermarks), and its counts replace the stream's. This is lambda's good idea, kept small: a correction, not a second pipeline of logic.
• Why does the sink upsert? The window result can update (allowed lateness) and can replay (recovery). Writing (ad_id, minute) → count as an overwrite makes both harmless; an increment would double count.

Dry run: one window, seven events, a 10 s bound and 60 s of allowed lateness

Ad A, tumbling windows of one minute, watermark = latest event time seen minus 10 s (the one-millisecond offset in Flink's example is ignored here), allowed lateness 60 s. Window W is [12:00:00, 12:01:00). The cleanup time for W is its end plus the lateness, 12:02:00.

Count the answers. The stream's final value for W is 4; the true count from event time is 5 (e1, e2, e3, e6, e7). The nightly batch recount reads the archive, finds all five and overwrites the row with 5, which is what billing sees. Skew values (arrival minus event time): e3 is 12 s, e6 is 45 s, e7 is 150 s. With a 10 s bound and no allowed lateness, e6 and e7 would both have been dropped, and W would have fired once with 3; allowed lateness recovers e6, and only e7, at 150 s of skew, needs the side output and the batch. With a 150 s bound every result would have waited two and a half minutes. The bound, the allowed lateness and the reconciliation are three settings on one trade-off.

Recognizing it in an unseen problem

• Signals: "per minute", "real time", "top K in the last hour", "fraud detection", "metrics", "click counts", "keep the search index in sync", "rebuild a view", or any requirement that mixes a latency target with a correctness target.
• First question: is the input bounded? A bounded job wants batch; an unbounded one wants a stream, and then you must say what time you use and when a result is final.
• If counts feed money or compliance: event time, deduplication, an idempotent sink and a batch reconciliation. If a few lost samples are harmless (telemetry), processing time with at-most-once is fine and far cheaper.
• If two stores must agree: do not dual write. Capture the change log, or use an outbox, and make each consumer idempotent.
• If the logic will change: keep the log long enough to replay, and be able to start a new job instance and cut over, not run two codebases.
• The pitfalls to name unprompted: the stalled watermark from an idle partition, the hot key, state that grows without bound (windows or dedupe sets that are never cleaned), and exactly-once claims that stop at the first external API call.
• The follow-up you will get: "what about events that arrive an hour late?" The answer is a policy, not a wish: bound, allowed lateness, side output, reconciliation, and a stated tolerance.