Loading...
Loading...
One Lua script holds 100 requests per minute exact across 12 gateways, yet every bucket meters events one request at a time. Counting words in 10TB by shuffling data to one machine takes days and blows memory. MapReduce, meaning a framework where many mappers run beside HDFS blocks plus reducers aggregate by key, colocalizes computation: 80k mappers read local 128MB blocks in parallel, emit sorted key-value pairs, then reducers pull partitions over HTTP and aggregate. Like a restaurant chain counting receipts in each branch before phoning totals to headquarters, you write only map and reduce while the framework handles placement, shuffle, and retries.
Input (HDFS 10TB, 80k blocks) → Map (80k mappers, data-local)
→ Map spills sorted buffer → Shuffle (partition by hash(key), sort, merge)
→ Reduce (pull via HTTP, merge, apply) → Output (HDFS)The naive single-machine loop reads, groups, and sums in one memory space, which fails when input exceeds RAM. MapReduce splits the contract three ways with explicit spill and sort between them. Map emits key-value pairs and spills sorted runs to disk; shuffle partitions by hash, sorts, and merges across the network; reduce merges sorted runs and commits via atomic rename so retries never publish halves.
Input reader splits file into records. Map of (k1,v1) to list(k2,v2) spills sorted buffers to disk, optionally combining locally to cut shuffle bytes.
def map(doc_id, text):
for w in text.split():
emit(w, 1)
# combiner (local reduce) runs before shuffle
# → (the, [1,1,1]) → (the, 3) locallyThe expensive stage partitions by hash(k2) modulo R, sorts by k2, merges, and fetches via HTTP to reducers. Skew, meaning one hot key holding half the data, stalls one reducer while siblings idle, which salting or a custom partitioner must relieve.
Reduce of (k2, list(v2)) merges sorted runs and writes committed output. Output commits via atomic rename to avoid partial visibility on retry.
def reduce(word, counts):
emit(word, sum(counts))Machine failure mid-job cannot mean rerunning 10TB. MapReduce retries the failed task attempt on another node, runs speculative duplicates only for the straggler tail, and treats mapper output as ephemeral until reducers fetch it. Classic Hadoop's single master was fatal on crash; YARN ResourceManager high availability, meaning standby masters coordinated by ZooKeeper-style election, now covers it. Iterative ML such as k-means still suffers, since each iteration is a full job with HDFS materialization between, which is why Spark caches datasets in memory instead.
Word-count over 10TB emitting (word, 1) per token produces roughly 1TB of map output. Every byte crosses to reducers, sorts, and merges. On a 10Gbps fabric with 40 reducers each pulls about 25GB plus external sort, taking minutes, while one skewed key such as a common word parking 50GB on one reducer leaves 39 siblings waiting. The naive job pays full shuffle; the tuned job combines map-side first.
10TB input, 80k mappers → 1TB map output, R=40 reducers: per reducer: ~25GB fetch + external sort + merge; minutes each combiner (the,[1,1,1]→3) cuts map output 10–100x for associative ops skew: key "the" = 5% of tokens → one reducer gets 50GB, rest 24GB → straggler Fixes that ship: map-side combine for sum/count (associative + commutative only) salting: emit (hotkey, rand(10)) → 10 sub-partitions, then second-stage aggregate custom partitioner: route hot keys by range, not hash % R
Each MapReduce job reads from HDFS and writes back, so ten iterations over 10TB move 100TB of disk I/O to nudge centroids. The write-read barrier between jobs plus a mandatory distributed sort of all map output, even when reduce is a plain sum, is the hidden tax. Spark avoids the sort for hash aggregation and spills only under pressure; Flink pipelines records without materializing stages. That barrier handed iterative ML to in-memory engines.
MapReduce sorts all map output by key at O(n log n) even for sums nobody needed sorted. Spark hash-aggregates without sorting and spills only on pressure; Flink streams records stageless. Profiling one shuffle teaches that distributed sort is the bill.
Hadoop duplicates only the straggler tail, meaning tasks 20% plus behind median after a minute, kills the loser, and commits the winner atomically. Over-eager speculation doubles load on shared racks, so cap speculative tasks near 10% of slots and disable for non-idempotent sinks.
Ten terabytes at 128MB blocks means 80,000 mappers whether you like it or not, since split size defaults to block size. Reducers are your one dial: too few sorts gigabytes each, too many drowns in HTTP fetches plus thousands of tiny outputs. Target just under full slot occupancy near 0.95 times reduce slots, overlap fetching with slow-start near 80% map completion, and run tiny jobs in uber mode, meaning inside the ApplicationMaster JVM with no containers.
Mappers: input bytes ÷ split size (default = block size 128MB) 10TB ÷ 128MB ≈ 80,000 maps; data-local, one wave per few thousand slots Reducers: R ≈ 0.95 × reduce slots (leave headroom for speculative duplicates) 40 slots → R=38; 1TB shuffle ÷ 38 ≈ 27GB sorted per reducer R=400 instead: 2.5GB each BUT 80k×400 fetch connections + 400 output files Small jobs: uber mode (run map+reduce in the ApplicationMaster JVM, no containers) Slow start: reducers start fetching at 5–80% map completion; 0.8 default overlaps well
Output-file warning: R reducers write R part-files. Ten thousand reducers for speed create ten thousand 100KB files, exporting the small-files problem to the next reader. Size R for the reader, not just the clock.
Clicks at 10TB joining users at 50GB on user_id tempts a full shuffle of both sides, sending 10TB across the wire. Broadcast join, meaning shipping the small side via distributed cache to every mapper, streams 10TB past a local 50GB table with zero shuffle while the small side fits mapper memory near 1 to 2GB. Sort-merge bucketed join, meaning pre-partitioning both sides by identical hash buckets once, merges bucket-to-bucket with no shuffle on every future join. Pay broadcast per job or bucketing once depending on repeat count.
Ship the 50GB user table via distributed cache to every mapper (fits in mapper RAM/disk), stream 10TB of clicks past it, emit joined rows with zero shuffle. Past a few GB per mapper, broadcast becomes the bottleneck it avoided.
Bucket both datasets by hash(user_id) into the same 256 buckets once; every future join merges bucket-to-bucket with no shuffle. Upfront cost, recurring payoff; the warehouse equivalent is clustered joins on the same key in BigQuery, Google's serverless warehouse, and Redshift, Amazon's warehouse.
Each mapper sorts in a 100MB buffer and spills at 80% while mapping continues into the remainder. A 2GB emit writes about 25 spills of 80MB, then merges in passes governed by merge factor 10, rereading and rewriting every pass. Each byte crosses disk roughly twice on spill plus twice on merge, so 2GB emit means 8GB of disk I/O. Raising the buffer to 256MB halves spills, raising merge factor to 32 to 64 cuts passes, and Snappy or LZ4 compression, both fast codecs, shrinks shuffle bytes 2 to 3x for 10% CPU.
Mapper emits 2GB, buffer 100MB, spill at 80% → ~25 spills of ~80MB merge factor 10: pass 1 merges 10+10+5 → 3 files; pass 2 merges 3 → 1 sorted run each byte written ~2x (spill + merge) and read ~2x; 2GB emit ≈ 8GB disk I/O combiner cuts emit 10x first: 200MB emit → 3 spills → single merge pass Knobs: mapreduce.task.io.sort.mb (100 → 256MB on big mappers halves spills) mapreduce.map.sort.spill.percent 0.8; mapreduce.task.io.sort.factor 10 → 32–64 compression (Snappy/LZ4) on map output: 2–3x less shuffle bytes for ~10% CPU.
Real pipelines chain ingest, clean, join, aggregate, export, where job 4 waits on job 3's _SUCCESS marker, meaning a zero-byte file signaling committed output. Schedulers such as Airflow, a workflow orchestrator, gate each stage on its predecessor, retry single failed stages, and check row counts plus null rates before poison spreads. Each stage writes a date-partitioned path atomically with unique attempt paths, so retrying stage 3 never duplicates stage 2. Late Tuesday events arriving Thursday recompute only Tuesday's downstream partitions by backfill parameter.
Each stage writes to a date-partitioned path and commits atomically, so retrying stage 3 never duplicates stage 2's output. Non-idempotent stages such as appending to one file or calling a webhook corrupt on every retry.
Tuesday's partition closed, then 2% more Tuesday events arrive Thursday. Recompute only Tuesday's downstream partitions, since partition-aligned idempotence makes backfill a date parameter. Unpartitioned outputs turn every late event into a full rerun.
Filtering 10TB of logs to 50GB of errors, projecting two columns from forty, or finishing counts with a map-side combine needs no grouping key. Map-only jobs with R equal to 0 write directly from mappers: no sort, no HTTP fetch, no reducer stragglers. The output is N mapper part-files the next stage coalesces from 80k to hundreds. DistributedCache, meaning Hadoop's broadcast file mechanism, carries a 2GB blocklist to every mapper with no join and no shuffle.
Map-only checklist: no grouping key needed? → R=0, skip shuffle entirely. 10TB → filter errors (0.5%) → 50GB across 80k part-files → coalesce to ~400 next stage reads 400 files, not 80k; the small-files tax paid once, deliberately. DistributedCache carries the 2GB blocklist to every mapper; no join, no shuffle.
A 1TB shuffle dominates job time, and Snappy or LZ4 compression, both fast splittable-friendly codecs, shrinks it 2 to 3x for 10% CPU: the cheapest speedup in Hadoop. Compress map output always, sequence files for storage, final output only when downstream readers agree on the codec. Gzip compresses smaller but decompresses slowly and stalls reducers; Bzip2 stays splittable but slow. Uncompressed shuffle is a choice to defend, never a default to keep.
A reducer failing at 90% merge completion tempts a full job retry, which replays 80,000 mappers for one task's bad luck. Rejected alternative: rerun the whole job so all 80,000 mappers recompute 10TB to replace one reducer's 25GB of fetched partitions, paying hours for a failure that touched one task. The framework instead replays only the failed reduce attempt: map outputs persist on mapper local disks until the job commits, so the replacement reducer re-fetches the same partitions over HTTP without recomputing maps. The edge that breaks this is mapper-disk expiry under long jobs, where completed map outputs age out before slow reducers fetch them, forcing selective map re-execution. Size reducer counts and slow-start overlap so fetches finish while mapper disks still hold the data, and treat repeated fetch-failures as a capacity signal rather than a network blip.
Reducer R7 dies at 90% merge (22GB of 25GB fetched, sorted runs on its disk):
retry: new attempt R7b on another node → re-fetches 40 mapper partitions via HTTP
mappers do NOT rerun; their spill files persist on mapper local disks until commit
expiry edge: map outputs garbage-collected after ~24h or disk pressure → if R7b
starts after expiry, its missing partitions trigger targeted map re-runs only.
slow-start 0.8 overlaps fetch with mapping so data is fresh when reducers need it.Eighty thousand mappers sort 1TB across the wire in minutes and reducer R7b re-fetches 25GB without rerunning one map. Yet the answer still lands when the whole job finishes. What blocks a fraudulent transaction in 300ms while the batch is still mapping?