Parallel &
Distributed Processing

Section 2.1 split storage into a single coordinator (the NameNode, metadata only) and many workers (DataNodes, holding actual blocks). Scheduling computation across a cluster uses the identical shape, for the identical reason: one place has to hold the global view, and it can't also be doing the heavy lifting, or it becomes the bottleneck it was supposed to prevent. The standard answer — YARN (Yet Another Resource Negotiator) — names four roles instead of two.

Worked Example · Launching "Count Orders Per Restaurant"

The Roles, Live

Practice

1. A cluster runs 4 simultaneous jobs. How many ApplicationMasters exist right now, and how many ResourceManagers?
   Solution

   4 ApplicationMasters (one per running job) and exactly 1 ResourceManager (one per cluster, regardless of how many jobs are running).

2. Block 2 (Section 2.1) lives on {Hub S, Hub E}. Both are fully busy with other containers when this job's AM asks for a data-local map container. What are the AM's realistic options?
   Solution

   Wait for a container to free up on Hub S or Hub E (preserving locality, at the cost of time), or accept a container on Hub N and have that map task fetch Block 2's data over the network (preserving speed-of-start, at the cost of Section 1.6's data-movement penalty). Real schedulers typically wait a short, configurable period before falling back to the second option.

3. The ApplicationMaster's own container dies partway through the job. Does the ResourceManager know which of the job's 3 map tasks had already finished?
   Solution

   No — the ResourceManager never tracked task-level state at all; only the AM did. This is precisely why AM failure is handled separately (the RM restarts the AM in a fresh container), and the restarted AM typically has to rediscover job progress from durable job-history logs, not from anything the RM was keeping.

4. Connect this to Section 2.4: quorum replication kept the storage layer working through a node failure by having no single point of failure among data holders. Does YARN's design give the same guarantee to the ResourceManager itself?
   Solution

   Not by the basic architecture above — a single ResourceManager is a single point of failure for scheduling new work, exactly the concern Section 2.1 raised about a single NameNode. Production YARN deployments address this with a standby ResourceManager that can take over, which is the scheduling-layer equivalent of Section 2.1's replication fix for the NameNode, not a difference in kind.

The job: count how many orders each restaurant received. Three map tasks, one per block, placed exactly as Section 3.1 described — each on a NodeManager that already holds its input.

Worked Example · Every Pair, Every Hop

Step Through It

Practice

1. If the job used only 2 reducers instead of 3, and Hub N took {Curry Point, Momo Hut, Wrap City} while Hub S took {Sushi Go, Pizza Barn}, how many shuffled records would each reducer handle?
   Solution

   Hub N: 4+3+3=10 records. Hub S: 3+3=6 records. Total 16, matching the job's full record count, just split differently — and more unevenly than the 3-reducer version in the worked example.

2. Which phase would a network outage between Hub N and Hub S actually block: map, shuffle, or reduce?
   Solution

   Shuffle specifically — it's the only phase that requires inter-node communication. Map tasks already running would finish untouched (purely local), and reduce tasks could still process any shuffle data that had already arrived before the outage.

3. Section 2.6 measured that a hub-selective query costs 1 network hop under hub-partitioning versus 3 under hash-partitioning. Does the reducer assignment in this section's worked example (alphabetical, not hash-based) change that lesson?
   Solution

   No — it's a different partitioning question entirely. Section 2.6 partitioned stored rows by a query-relevant key; this section partitions shuffle keys (restaurant names) across reducers for a computation already in flight. Both are partitioning decisions, but one is about where data rests and the other is about where a specific job's intermediate results land.

4. Hub E's reducer, handling only Wrap City, finishes well before Hub N's. Is Hub E's NodeManager now free to pick up unrelated work from a different job?
   Solution

   Yes — once its container's task completes, the NodeManager reports the freed resources to the ResourceManager (Section 3.1), which can immediately grant that capacity to any other job waiting for containers. An idle reducer doesn't sit reserved for its original job "just in case."

A combiner runs the reduce function's logic locally, before shuffle, on whatever a single map task already produced. It's not a new phase — it's the reduce logic borrowed early, applied to one map task's own output, on that same machine, with no network involved.

Worked Example · Block 1, With and Without

One pair saved, from sixteen rows, looks unimpressive. It isn't the point at this scale — it's the mechanism, and the mechanism's payoff is proportional to how many repeats of the same key land on the same map task, which grows directly with data volume.

The Same Mechanism, at Citywide Volume

The combiner didn't change the final answer by one order — Section 3.2's reduce step still adds up to the identical totals either way. It changed how many bytes had to move to get there, which is exactly Section 1.6's "moving computation to data" principle, applied one level deeper: instead of moving the whole computation, move just enough of it — the associative part — to happen before the expensive hop.

Practice

1. Block 2's local counts (Section 3.2) are Momo Hut:1, Wrap City:1, Curry Point:1, Pizza Barn:1, Sushi Go:1. Does a combiner save any pairs here? Why or why not?
   Solution

   No savings — every restaurant in Block 2 appears exactly once, so pre-summing changes nothing (each pair is already a "sum of one"). Combiners only save shuffle traffic when the same key repeats within one map task's own output, which happened for Curry Point in Block 1 but not for anything in Block 2.

2. Could a combiner be written for "find the maximum order amount per restaurant"? Sketch why or why not.
   Solution

   Yes — max is associative and commutative exactly like sum: max(max(a,b),c) = max(a,b,c) regrouped any way. A map task seeing two Curry Point orders locally (₹340 and ₹310) can combine them to a single (Curry Point, 340) pair before shuffle, since 340 already IS the local maximum.

3. A job computes "average order amount per restaurant" with a naive (broken) combiner that averages locally and ships (restaurant, local_average) pairs. Curry Point's four orders are ₹340 and ₹310 in Block 1, ₹355 in Block 2, and ₹330 in Block 3. Show the naive combiner gives the wrong final answer.
   Solution

   True average across all four: (340+310+355+330)/4 = 1335/4 = ₹333.75. The naive combiner ships three local averages instead — Block 1's (340+310)/2=325, Block 2's 355, Block 3's 330 — and the reducer averages THOSE three numbers: (325+355+330)/3 = ₹336.67. Wrong, because it silently treated Block 1's two real orders as worth the same as Block 2's and Block 3's one each, overweighting the smaller blocks.

4. Redo the same computation with the correct (sum, count) combiner instead. Confirm it recovers the true ₹333.75.
   Solution

   Block 1 ships (650, 2), Block 2 ships (355, 1), Block 3 ships (330, 1). The reducer sums both fields separately: sum = 650+355+330 = 1335, count = 2+1+1 = 4, then divides exactly once: 1335/4 = ₹333.75 — matching the true average, because carrying the count forward preserves how many real orders each partial sum actually represents.

Section 3.2's reducer assignment already split unevenly — 7 shuffled records to Hub N, 6 to Hub S, only 3 to Hub E — simply because two busier restaurants happened to land on the same reducer. That was a mild version of a problem called data skew: when the real-world distribution of a key isn't uniform, no partitioning scheme that just spreads keys evenly can guarantee it spreads work evenly, because some keys carry far more records than others.

Worked Example · A Chain Restaurant, at Citywide Scale

Sixteen rows can't show this convincingly — the whole point of skew is that it hides at small scale and bites at large scale. Citywide, CityCourier serves roughly 1,200 distinct restaurants across 46,080 orders/day.

Every other reducer in the job finishes in the time it takes to process a few dozen records. Curry Point's reducer is still working through 5,530 — and because a MapReduce job doesn't finish until its last task finishes, the entire job's runtime is now set by that one straggler, no matter how fast the other 1,199 reducers were.

Two Real Fixes

Practice

1. If Curry Point's share were 25% instead of 12%, recompute its daily order count and the skew ratio against the 38.4/day average.
   Solution

   46,080 × 0.25 = 11,520 orders/day. Skew ratio: 11,520 ÷ 38.4 = 300× the average restaurant's load — roughly double the worked example's 144×, since the share itself roughly doubled.

2. With salting into 10 sub-keys, roughly how many orders does each of Curry Point's 10 reducer partitions handle, and how does that compare to the citywide average restaurant's 38.4/day?
   Solution

   5,530 ÷ 10 ≈ 553 orders per sub-key — still about 14× the average restaurant's load (553/38.4), far better than 144× unsalted, but not perfectly even. Salting reduces skew; it doesn't guarantee eliminating it, since the choice of how many sub-keys to split into is itself a tuning decision.

3. A job joins CityCourier's orders against a restaurants table, keyed by restaurant name. Explain why Curry Point being a skewed key is now a problem for the join, not just for aggregation.
   Solution

   A join has to bring together every order row AND the matching restaurant row for the same key on one machine to actually perform the join. Since every one of Curry Point's 5,530 daily order rows shares that same key, all 5,530 land on whichever machine is handling the Curry Point side of the join — identical mechanism to the reduce-side skew above, just triggered by a join key instead of a group-by key.

4. Section 1.5 showed a 2-node split (11 vs 5) being less efficient than a balanced 3-node split, due to hardware imbalance. How is Curry Point's skew fundamentally different from that earlier imbalance?
   Solution

   Section 1.5's imbalance was a scheduling choice — the same 16 orders could have been split more evenly across nodes, and a better split would have fixed it entirely. Curry Point's skew is a property of the data itself: no matter how the job assigns keys to reducers, every one of Curry Point's orders must still end up together on one reducer for a correct count, so re-splitting the assignment can spread OTHER keys better but can never split Curry Point's own total across multiple reducers without an extra mechanism like salting.

MapReduce's contract is rigid on purpose: one map phase, one shuffle, one reduce, done. A computation with two logically dependent steps has to run as two separate MapReduce jobs, the first writing its output to HDFS, the second reading that output back in. Spark's answer is to represent the whole computation — however many steps — as one DAG (directed acyclic graph) of operations, and let the scheduler decide how much of it can run without ever touching disk in between.

Worked Example · One DAG vs Two Jobs

Reuse Section 3.2's job — count per restaurant — then add the second step: keep only restaurants with a count of 4 or more. Only Curry Point (count 4) survives; the other four restaurants (all count 3) are dropped.

See the Graph

Practice

1. Add a third step: after filtering to count≥4, sort the surviving restaurants by name. Does this add a new stage to the Spark DAG?
   Solution

   Sorting by key generally requires a shuffle (records need to be redistributed by sort order across partitions), so yes — it adds a second shuffle and therefore a new stage boundary, giving 3 stages total instead of 2. Sorting is a wide-dependency operation just like grouping and joining.

2. Would a chain of filter → map → filter, with no grouping or joining anywhere, ever force a shuffle?
   Solution

   No — all three are narrow-dependency operations, each output partition depending on exactly one input partition. The entire chain pipelines into a single stage with zero shuffles, regardless of how many such operations are chained together.

3. A Spark job calls .count() on a chain of transformations, then calls .collect() on the exact same chain again, without caching anything in between. What actually happens, cost-wise?
   Solution

   The entire DAG — every transformation, including any shuffle — runs twice, once triggered by each action. Nothing from the first run is reused, exactly the "lazy evaluation's real trap" above; the fix is calling .cache() on the chain before either action, so the second action reuses the first run's materialized result instead of recomputing from source.

4. Section 3.1's ApplicationMaster tracked one job's tasks for the ResourceManager. In a Spark job with 2 stages, is that still one ApplicationMaster, or one per stage?
   Solution

   Still one — the ApplicationMaster is per job (or per Spark application, which can itself run many actions/DAGs across its lifetime), not per stage. The same AM negotiates containers for Stage 0's tasks, then for Stage 1's tasks once the shuffle they depend on is ready, exactly the same role Section 3.1 described, just coordinating a multi-stage plan instead of a single map-then-reduce one.

Both MapReduce and Spark answer the same way in spirit: retry a bounded unit of work, not the whole job. They differ in what that unit is anchored to, and that difference has real cost consequences.

Worked Example · Redoing Only What Was Lost

The Same Question for Spark

Spark's retry unit is a lost RDD partition, recomputed by replaying its lineage — the recorded chain of transformations that built it. If Hub N's reduce-side partition is lost, Spark needs the shuffle data the map stage already wrote to disk (same durable artifact MapReduce relies on) to recompute it — often just as cheap as the MapReduce retry above.

Practice

1. If Hub E's reduce task (handling only Wrap City, 3 records) fails instead of Hub N's, what fraction of the job gets redone?
   Solution

   3 of 16 records, or 18.75% — smaller than Hub N's 43.8%, simply because Hub E's reducer was handling fewer keys' worth of shuffled data. This is exactly Section 3.2's load-imbalance note showing up again: an uneven reducer assignment means failures aren't equally expensive to retry either.

2. A map task (not a reduce task) fails, before any of its output has been shuffled anywhere. What has to be redone, and does it touch the network?
   Solution

   Just that one map task, re-reading its own local block (Section 2.1) and re-emitting its pairs — no network involved, since map tasks only ever touch local data. This is the cheapest possible failure: nothing had been shuffled yet, so nothing downstream needs to change.

3. Chapter 2's replication factor 2 meant a DataNode failure never lost a block outright. Does an equivalent guarantee exist for a failed MAP task's output, once it's been shuffled to a reducer?
   Solution

   Generally yes in practice: shuffle output is written to local disk (not held only in the map task's memory) precisely so a downstream reduce failure doesn't force map tasks to rerun. But this is weaker than Section 2.1's replication — that shuffle file usually exists on ONE machine's local disk, not replicated 2× across machines, so losing that specific machine before the reducer reads it would force the map task to rerun after all.

4. Tie this section back to Section 3.1's ApplicationMaster. What specifically does the AM need to know to decide a task has failed, and where does that information come from?
   Solution

   Missing heartbeats or progress reports from the task's container, over some timeout window — the exact same signal Section 2.1's NameNode uses to detect a dead DataNode, just scoped to one running task instead of one storage node. Neither system needs a human to notice; both rely on the absence of an expected periodic signal.

1. A cluster's ResourceManager crashes. Can already-running tasks keep executing?
   Solution

   Yes — the RM's job is granting NEW containers, not supervising tasks already running (that's each job's own ApplicationMaster). Already-running containers keep executing; only the ability to start new work stalls until the RM (or a standby) is back.

2. A map task's output never gets used by any reducer, because the job is killed before shuffle starts. Was running that map task wasted work?
   Solution

   In terms of final output, yes — nothing downstream consumed it. But it cost nothing beyond that one task's local compute time; no network resources were spent, since shuffle (the networked phase) never began.

3. Explain, in one sentence, why "average" needs a different combiner strategy than "sum" or "count."
   Solution

   Sum and count are directly additive across partial results, but a partial average discards the count that produced it, so combining partial averages naively double-weights whichever partition had fewer records — the fix is carrying (sum, count) forward and dividing only once, at the end.

4. A hash-partitioned job sends every instance of key "Curry Point" to reducer #2 by design. Is that a bug?
   Solution

   No — it's required for correctness. Every record for the same key MUST land on the same reducer for a group-by or join to produce a right answer. The problem in Section 3.4 wasn't that Curry Point's records were grouped together; it's that the volume behind that one key was disproportionate.

5. A Spark job chains 6 map calls, then one reduceByKey, then 2 more map calls. How many stages does this DAG have?
   Solution

   2 stages. All narrow-dependency operations (every map) pipeline together with whatever's adjacent; only the single reduceByKey forces a shuffle, creating exactly one stage boundary regardless of how many map calls sit on either side of it.

6. Which is cheaper to retry after a failure: a map task that hasn't shuffled yet, or a reduce task that has already pulled its shuffle data?
   Solution

   Both retries are bounded to their own task, but the map task's retry is purely local (re-read its block, no network); the reduce task's retry still needs to re-pull its shuffle partition from wherever the map outputs are stored, which may or may not involve the network depending on where that data sits relative to the retried task's new container.

7. Why can't naive hash-partitioning fix Curry Point's skew just by using more reducers?
   Solution

   Because standard hash-partitioning sends every occurrence of one key to exactly one partition, however many partitions exist. Adding reducers spreads OTHER keys thinner but does nothing to split Curry Point's own single-key load, which is why Section 3.4 needed salting (splitting the key itself) rather than just adding capacity.

8. A cached Spark result and a Section 2.1 replicated HDFS block both exist so something doesn't have to be recomputed or refetched from scratch. What's the key difference in what happens if the machine holding each one dies?
   Solution

   A replicated block (RF≥2) survives the failure outright — another copy already exists elsewhere. A cache with no replication is simply gone, and Spark must fall back to recomputing it via lineage. Caching speeds up repeated access; it isn't, by itself, a durability mechanism the way replication is.

• Hadoop: The Definitive Guide — Tom White. Covers YARN's ResourceManager/NodeManager/ApplicationMaster split (Section 3.1) and MapReduce's job execution internals (Section 3.2) in full implementation detail.
• Spark: The Definitive Guide — Chambers & Zaharia. The authoritative source on RDDs, the DAG scheduler, narrow versus wide dependencies, and lineage-based fault tolerance behind Sections 3.5 and 3.6 — essential before Chapter 4's deeper dive into in-memory processing.
• Mining of Massive Datasets — Leskovec, Rajaraman & Ullman. Its treatment of combiners and skewed-key handling in MapReduce (Sections 3.3–3.4) predates Spark but the underlying shuffle-cost reasoning is identical.