Batch Processing Algorithms

Batch engines schedule a DAG (or a linear two-phase MapReduce job) over partitions. Embarrassingly parallel map/filter stages need little coordination. Group-by, join, and global sort force a shuffle: data is repartitioned by key, often spilled through external sort. That shuffle is where cost and skew live.

Joins pick an algorithm from sizes and memory: broadcast hash join, partitioned/grace hash join, or sort-merge join. Combiners and partial aggregations shrink bytes before they cross the network. Failures replay from durable map outputs or from lineage; explicit checkpoints trade rewrite cost for shorter recovery.

The product promise is different from streaming: you optimize for correct completion by a deadline (SLA), not for millisecond watermarks. Micro-batches blur the line; the same algorithms still sit under the stage planner.

When intermediate data exceeds RAM, engines sort with replacement-selection or quicksort-sized runs, write runs to disk, then k-way merge. Network shuffle often lands unsorted chunks that must be sorted before a streaming reduce can assume key order. That is why “CPU is fine but the job is slow” often means disk-bound merge.

External sort sketch:
  while input remains:
    fill memory with records
    sort → write run_i to disk
  merge runs with a k-way heap (or multi-pass if too many runs)

I/O lower bound: roughly 2 passes over data for sort
  (write runs + read for merge), more if merge is multi-level

Design knobs:
  more memory → fewer/larger runs → cheaper merge
  more spill disks / better local SSDs → faster shuffle
  fewer giant partitions → less per-task sort pressure

A combiner is a map-local reduce. Sums, counts, min/max, and sketch merges (HLL union) are safe. Average is not a safe one-shot combine unless you emit (sum, count). Distinct counts need sketches or a two-phase unique-then-count. Getting this wrong silently returns plausible wrong numbers—the worst batch failure mode.

Safe combine:     sum, count, min, max, HLL.merge
Unsafe one-shot:  mean, median, distinct count (exact)
Pattern for mean:
  map → emit (key, sum=x, count=1)
  combine/reduce → sum sums, sum counts
  final → sum/count

Two-phase distinct:
  stage1: map unique (key, user) → shuffle → distinct users per key
  stage2: count per key
  or: use HyperLogLog merge in one stage if approx OK

Stages are barrier-shaped: the slowest task defines finish time. A straggler from key skew (one user_id with 5% of events) holds everyone. Mitigations: better partitioners, key salting with fan-in, range repartition from sampled quantiles, speculative execution of slow tasks, and isolating hot keys into dedicated reducers.

Also watch input skew: a few huge files versus millions of tiny ones. Compact small files in a prior batch; split large ones. Scheduling overhead and namenode/listing cost are real batch algorithms problems, not just “ops.”

Classic MapReduce materializes map output. Spark’s lineage recomputes lost partitions until you checkpoint or persist at a wide dependency. Warehouse tables materialize each CREATE TABLE AS SELECT. Choose based on recomputation cost versus storage and on how often nodes die mid-flight.

Recovery cost sketch:
  Job 4 hours, failure at 3:50 without checkpoint
    lineage recompute of expensive shuffle ≈ hours
  Checkpoint after heavy shuffle (~30 min write):
    failure at 3:50 → replay ~10 min of tail stages

Rule:
  checkpoint after expensive shuffles you refuse to redo
  do not checkpoint every tiny narrow transform

Exactly-once batch outputs:
  write to temp + commit (manifest rename / table transaction)
  never append partially visible files as “done”