MapReduce is the classic batch-processing abstraction for turning a large, partitioned data set into another large, partitioned data set without forcing the application developer to hand-code scheduling, retries, network transfers, and failure recovery. The original Google paper describes a deliberately small interface: a map function emits intermediate key/value pairs, a reduce function receives all values for one intermediate key, and the runtime handles partitioning, shuffle, sort, remote reads, and task re-execution across thousands of commodity machines.
The idea matters beyond the historical Hadoop ecosystem. Spark, Flink batch mode, Beam, SQL-on-object-store engines, and cloud warehouses all preserve the same physical pattern: scan partitions in parallel, repartition records by a key, sort or hash-group data at a boundary, then run an aggregation, join, or write step. DDIA uses MapReduce as the entry point for understanding batch jobs because it exposes the expensive part of distributed data processing: moving data to the place where records with the same key can meet.
Distributed joins are the most important example. A local database joins tables inside one process with nested loops, hash joins, or sort-merge joins. A distributed engine must first decide whether to move the small side to every worker, repartition both sides by join key, exploit pre-existing partitioning, or perform a multi-stage aggregation to reduce the amount of shuffled data. The join algorithm is therefore an architecture decision, not just a query-plan detail.
Good batch systems are boring on purpose. They assume machines fail, slow disks happen, some records are malformed, and a few partitions are much larger than the rest. They make deterministic tasks restartable, write outputs atomically, keep lineage or checkpoint metadata, and expose counters so operators can prove the job did what the job author intended.
The map phase is embarrassingly parallel: each task reads an input split and emits zero or more intermediate records. The shuffle phase partitions those intermediate records by key, usually with hash(key) mod R, where R is the number of reduce partitions. The sort/group phase makes all records for one key adjacent. The reduce phase consumes a key and its values and emits final records.
The original MapReduce implementation was tuned for clusters with local disks and limited bisection bandwidth. Map tasks ran near their input blocks when possible, wrote intermediate files to local disk, and let reduce workers fetch those files remotely. If a map worker failed, its intermediate files disappeared, so the map task was re-executed. If a reduce worker failed after writing to a replicated file system atomically, its final output did not need to be recomputed.
A distributed join asks where matching records should meet. In a repartition join, both inputs are shuffled by the join key so every worker owns a disjoint key range. In a broadcast join, the smaller input is copied to all workers and each worker joins it with its local slice of the large input. In a bucketed or colocated join, both inputs are already partitioned compatibly, so the engine avoids a full shuffle.
The expensive boundary in a distributed batch job is usually not the user function. It is materializing intermediate data, transferring it across the network, sorting or hashing it, and recovering when one producer or consumer fails. A one-line groupByKey in an API can imply terabytes of remote reads and writes.
MapReduce can re-run a task after a worker dies because deterministic map and reduce functions produce the same logical output for the same input. Non-deterministic functions, external side effects, and time-dependent output weaken this property and make retries observable.
A batch runtime breaks the input into many more tasks than machines, schedules map tasks, tracks intermediate file locations, schedules reduce tasks, and commits final outputs. The master or driver is mostly metadata: task state, worker leases, split assignment, file locations, counters, and commit records.
type Record struct {
Key string
Value []byte
}
func RunMapReduce(inputSplits []Split, reducers int) error {
mapOutputs := make(chan MapOutput)
for _, split := range inputSplits {
go func(s Split) {
tmp := newPartitionedWriter(reducers)
for rec := range s.Read() {
for _, out := range mapFn(rec) {
p := hash(out.Key) % reducers
tmp.Write(p, out)
}
}
mapOutputs <- tmp.CommitLocalFiles()
}(split)
}
locations := collectMapOutputs(len(inputSplits), mapOutputs)
for r := 0; r < reducers; r++ {
go func(partition int) {
files := fetchPartitionFiles(locations, partition)
sorted := externalSort(files, func(a, b Record) bool {
return a.Key < b.Key
})
out := newAtomicOutput(partition)
for key, values := range grouped(sorted) {
for _, result := range reduceFn(key, values) {
out.Write(result)
}
}
out.Commit()
}(r)
}
return waitForAllReducers()
}A reduce-side join tags records from each input, emits them by join key, and joins grouped values in the reducer. It works for large-large joins because both sides are partitioned by the key. Its cost is that both sides cross the shuffle boundary unless they are already bucketed compatibly.
func mapCustomer(c Customer) []Record {
return []Record{{
Key: c.ID,
Value: encode(Tagged{Side: "customer", Payload: c}),
}}
}
func mapOrder(o Order) []Record {
return []Record{{
Key: o.CustomerID,
Value: encode(Tagged{Side: "order", Payload: o}),
}}
}
func reduceJoin(customerID string, values Iterator[Tagged]) []JoinedOrder {
var customer *Customer
orders := make([]Order, 0)
for values.Next() {
v := values.Value()
switch v.Side {
case "customer":
c := v.Payload.(Customer)
customer = &c
case "order":
orders = append(orders, v.Payload.(Order))
}
}
if customer == nil {
return nil
}
out := make([]JoinedOrder, 0, len(orders))
for _, order := range orders {
out = append(out, JoinedOrder{Customer: *customer, Order: order})
}
return out
}A broadcast join copies a small dimension table to every map worker. The large table is scanned without repartitioning, and each worker probes an in-memory hash table. This avoids a shuffle of the large side, but it only works when the broadcast side fits comfortably in worker memory and can be distributed reliably.
Inverted index construction maps documents to (term, document) pairs, shuffles by term, sorts document IDs, and writes posting lists. This is a natural MapReduce workload because the reduce function owns one term at a time and can produce a compact, ordered posting list.
Data warehouses routinely join clickstream, order, or telemetry facts against smaller user, product, or account dimensions. Broadcast joins are often best when a dimension is small and slow-changing. Repartition joins are required when both sides are large or when the dimension table is too large to fit on every executor.
Batch jobs remain the safest way to rebuild derived state after a bug. A streaming pipeline may update an index continuously, but the batch job can recompute the entire index from source-of-truth logs, compare counts, and replace the output atomically when the result passes validation.
A combiner performs local aggregation on each mapper before data is shuffled. It is valid only when the operation is associative and commutative, such as sums, counts, minimums, and maximums. It is not valid for arbitrary averages unless it emits count and sum as an aggregate state.
type Count struct {
Sum int64
}
func combineWordCounts(records []Record) []Record {
counts := map[string]int64{}
for _, r := range records {
counts[r.Key] += decodeInt(r.Value)
}
out := make([]Record, 0, len(counts))
for word, count := range counts {
out = append(out, Record{Key: word, Value: encodeInt(count)})
}
return out
}Hot keys can overload one reducer. A common mitigation is salting: split one hot key into several salted subkeys, duplicate the small side across those salts, and merge the results. Salting trades extra work and duplication for lower tail latency.
func saltedOrderKey(customerID string, orderID string, saltBuckets int) string {
if isHotCustomer(customerID) {
return customerID + "#" + strconv.Itoa(hash(orderID)%saltBuckets)
}
return customerID + "#0"
}
func saltedCustomerKeys(customer Customer, saltBuckets int) []string {
if !isHotCustomer(customer.ID) {
return []string{customer.ID + "#0"}
}
keys := make([]string, 0, saltBuckets)
for i := 0; i < saltBuckets; i++ {
keys = append(keys, customer.ID+"#"+strconv.Itoa(i))
}
return keys
}Tasks should write to attempt-specific temporary paths and commit by atomic rename or manifest update. This prevents duplicate attempts from exposing partial output. Modern object stores complicate rename semantics, so many engines use commit protocols that publish a manifest after all task files are durable.