Documentation Index
Fetch the complete documentation index at: https://resources.devweekends.com/llms.txt
Use this file to discover all available pages before exploring further.
Chapter 6: Fault Tolerance and Reliability
In distributed systems, failure is not an exception — it is the norm. Hadoop was designed from the ground up to operate reliably in environments where individual components fail regularly. This chapter explores the sophisticated fault tolerance mechanisms that make Hadoop one of the most resilient distributed systems ever built. The intellectual foundation for Hadoop’s fault tolerance comes from two sources. First, the Google File System paper (2003), which reported that Google’s clusters experienced roughly 1,000 individual disk failures and dozens of machine failures per day across their fleet. Google’s engineers did not try to prevent these failures — they designed a system that treated them as routine events. Second, Jim Gray’s landmark work on fault-tolerant computing at Tandem Computers (1980s), which established the principle that hardware will fail and software must compensate. The Hadoop project inherited both philosophies: replicate data, retry failed computations, and never trust any single machine. What makes Hadoop’s approach distinctive is the separation of concerns. HDFS handles data durability (your data survives hardware failures), MapReduce handles computation reliability (your job completes despite task failures), and YARN handles resource availability (the cluster continues scheduling work despite node losses). Each layer has its own fault tolerance mechanisms, and they compose together to create a system where the end user rarely sees the failures that are constantly happening underneath.Chapter Goals:
- Master HDFS replication and recovery mechanisms
- Understand MapReduce fault tolerance strategies
- Learn YARN reliability features
- Explore failure detection and handling
- Design fault-tolerant Hadoop applications
The Philosophy of Failure in Distributed Systems
The intellectual roots of Hadoop’s fault tolerance trace back to a single, uncomfortable truth that Jim Gray articulated in his 1985 paper “Why Do Computers Stop and What Can Be Done About It?”: hardware fails, software has bugs, and operators make mistakes. The question is never whether a component will fail, but when and how many at once. Google internalized this reality when building GFS and MapReduce in the early 2000s. Their commodity hardware clusters experienced disk failures daily, node reboots weekly, and entire rack outages quarterly. The original GFS paper (2003) explicitly states that “component failures are the norm rather than the exception.” Hadoop inherited this philosophy wholesale. Every design decision — from triple replication to speculative execution — flows from the assumption that individual components are unreliable, and reliability must be an emergent property of the system as a whole. This is the same principle that underpins modern cloud-native systems: AWS designs every service to tolerate Availability Zone failures, Kubernetes restarts crashed pods automatically, and Cassandra replicates across data centers. Hadoop was arguably the first open-source system to make this philosophy accessible to the entire industry.Why Fault Tolerance is Critical
Types of Failures Hadoop Handles
Hardware Failures
Hardware Failures
Disk Failures:
- Individual disk corruption or crash
- RAID controller failures
- Storage network issues
- Silent data corruption
- Complete server crash
- Power supply failure
- Memory errors (ECC failures)
- CPU/motherboard failure
- Network interface card failure
- Switch failures
- Cable disconnections
- Network partition (split-brain)
- Bandwidth congestion
- DNS resolution failures
Software Failures
Software Failures
Process Crashes:
- Out of memory errors
- Segmentation faults
- Uncaught exceptions
- Resource exhaustion
- Data processing errors
- Infinite loops
- Deadlocks
- Race conditions
- Incorrect settings
- Version mismatches
- Permission issues
Performance Degradation
Performance Degradation
Slow Nodes (Stragglers):
- Disk degradation
- CPU thermal throttling
- Memory pressure
- Network congestion
- Competing workloads
- Background processes
- I/O bottlenecks
HDFS Fault Tolerance Mechanisms
Data Replication Architecture
Deep Dive: Rack Awareness vs. DynamoDB AZ-Awareness
Hadoop’s rack-aware placement policy was not the first system to distribute replicas across failure domains, but it was one of the first to make the topology explicitly configurable by operators. The concept maps directly to the “failure domain” abstraction used across distributed systems. AWS Availability Zones, Azure Fault Domains, and GCP Zones all implement the same fundamental idea: ensure that a single blast radius event (power failure, network partition, cooling system outage) cannot destroy all copies of your data. Understanding this pattern is essential because interviewers will expect you to draw connections between Hadoop’s rack awareness and how modern cloud-native databases like DynamoDB, Cosmos DB, and Cloud Spanner handle the same problem automatically. Hadoop’s Rack Awareness and DynamoDB’s AZ-Awareness share the same goal: ensuring that a single infrastructure failure (like a top-of-rack switch or a data center power outage) does not result in data loss.1. The Strategy Comparison
| Feature | Hadoop Rack Awareness | DynamoDB AZ-Awareness |
|---|---|---|
| Fault Domain | Physical Server Rack (Power/Network) | AWS Availability Zone (Data Center) |
| Data Placement | 1st: Local, 2nd: Remote Rack, 3rd: Same as 2nd | Automatically spread across 3 AZs |
| Distance Metric | Network hops (Topology script) | Logical AZ IDs |
| Consistency | Eventual (Replication Pipeline) | Strong or Eventual (Paxos Quorum) |
2. Implementation Differences
- Hadoop: Relies on a Topology Script (typically
/etc/hadoop/conf/topology.sh) that maps IP addresses to rack IDs (e.g.,/dc1/rack1). The NameNode uses this map to calculate the “distance” between nodes. - DynamoDB: Managed by AWS. The partitioning logic automatically ensures that replicas for a single partition are never placed in the same AZ. This is transparent to the user, unlike Hadoop where you must manually configure the topology.
3. Locality vs. Durability
- Hadoop Locality: Prioritizes “Local Reads” (Reading from the same node). If the local node is down, it falls back to the same rack.
- DynamoDB Locality: Prioritizes “AZ Locality” for reads. While you can’t request a “Node Local” read, DynamoDB routes requests to the closest healthy replica to minimize latency.
Replication Factor Configuration
Deep Dive: Erasure Coding (Reed-Solomon) in Hadoop 3.x
Erasure coding is not a Hadoop invention — it has been used in RAID systems since the 1980s (RAID-5 and RAID-6 are special cases of erasure coding) and in telecommunications for decades before that. The Reed-Solomon algorithm itself dates back to a 1960 paper by Irving Reed and Gustave Solomon. What Hadoop 3.x did was bring erasure coding to the distributed file system layer, applying it across nodes and racks rather than across disks within a single machine. This was a significant engineering achievement because distributed erasure coding introduces network latency during reconstruction — a trade-off that does not exist in local RAID. Facebook (now Meta) was one of the earliest large-scale adopters of HDFS erasure coding through their HDFS-RAID project (circa 2010), which predated the official Hadoop 3.x implementation and saved them petabytes of storage. The same principle now powers cloud storage backends: Azure Storage uses erasure coding (Local Reconstruction Codes) internally, and Amazon S3’s eleven-nines durability guarantee relies on similar Reed-Solomon-based schemes spread across Availability Zones. One of the biggest limitations of standard 3-way replication is the 200% storage overhead. For every 1TB of data, you need 3TB of physical disk space. Hadoop 3.x introduced Erasure Coding (EC) to solve this, providing the same level of fault tolerance with significantly less storage.1. How Erasure Coding Works
Unlike replication, which copies the entire block, EC uses mathematical formulas (typically Reed-Solomon) to generate “parity” data.- Data Cells (d): The original data blocks.
- Parity Cells (p): Calculated from the data cells.
- Policy (d+p): A common policy is RS-6-3 (6 data blocks, 3 parity blocks).
2. Trade-offs: Durability vs. Performance
| Feature | 3x Replication | Erasure Coding (RS-6-3) |
|---|---|---|
| Storage Overhead | 200% | 50% |
| Fault Tolerance | Can lose 2 nodes | Can lose 3 nodes |
| Write Performance | High (Sequential I/O) | Lower (Requires parity calculation) |
| Read Performance | High (Local reads) | Lower (Requires network for reconstruction) |
| CPU Usage | Low | High (XOR and Galois Field math) |
| Best For | Hot data (frequent access) | Cold data (archives) |
3. Striped vs. Contiguous Layout
HDFS Erasure Coding uses a Striped Layout. Instead of coding across entire large blocks (128MB), it codes across small “cells” (typically 64KB or 1MB) within a block group. This allows for parallel I/O and reduces the “wait time” for reconstruction.The Reed-Solomon (RS-6-3) Mathematical Proof
To understand why RS-6-3 can lose 3 blocks, we look at the Vandermonde matrix used in Galois Fields ():- Data Vector ():
- Coding Matrix (): A matrix where the top is an Identity Matrix () and the bottom is the Generator Matrix ().
- Resulting Code ():
Block Recovery Process
4. The Block Lifecycle State Machine
The NameNode maintains a state machine for every block in the system. Understanding these transitions is key to debugging “missing” or “corrupt” data.- UNDER_REPLICATED: Replicas < configured replication factor ().
- PENDING: The NameNode has commanded a DataNode to copy the block, but the DataNode hasn’t reported back yet.
- HEALTHY: Replicas = .
- CORRUPT: A client or DataNode background scanner reported a checksum mismatch. The NameNode will NOT use this block for replication; it will only use healthy replicas to fix it.
NameNode HA Configuration
Checksum Verification
MapReduce Fault Tolerance
MapReduce’s fault tolerance model is fundamentally different from HDFS’s because compute failures are handled at the task level, not the data level. The original Google MapReduce paper (2004) reported that a typical MapReduce job at Google would experience multiple worker failures during execution, and the framework was designed to transparently re-execute failed tasks without any user intervention. This “re-execution” approach works because MapReduce tasks are stateless transformations of immutable input data — if a task fails, you simply run it again on the same input split. This is a much simpler model than the checkpoint-and-recover approach used by systems like MPI (Message Passing Interface), where a single node failure often requires restarting the entire computation from the last global checkpoint. The trade-off is that MapReduce sacrifices the ability to handle fine-grained, long-running stateful computations — a limitation that later frameworks like Apache Flink and Spark Streaming addressed with their own checkpointing mechanisms.Task Failure Handling
ApplicationMaster Failure Recovery
Speculative Execution
Speculative execution is Hadoop’s answer to the “straggler problem” — the observation that in a large cluster, a small number of tasks will run significantly slower than the rest due to hardware degradation, resource contention, or other unpredictable factors. Google’s original MapReduce paper identified stragglers as one of the most significant sources of job latency: a single slow task can hold up an entire job that is otherwise 99% complete. The idea of launching redundant work to hedge against slow responses is not unique to MapReduce. DNS resolution uses a similar technique (sending queries to multiple resolvers), and modern microservice architectures use “hedged requests” (sending the same RPC to two backends and taking the first response). Dean and Barroso’s influential 2013 paper “The Tail at Scale” formalized this pattern and showed that issuing redundant requests is often the most cost-effective way to reduce tail latency at scale. In practice, speculative execution in Hadoop trades a modest increase in cluster resource usage (typically 2-5%) for a significant reduction in job completion time.MapReduce Fault Tolerance Configuration
YARN Fault Tolerance
YARN’s fault tolerance represents a significant evolution over the Hadoop 1.x JobTracker model. In the original Hadoop architecture, the JobTracker was both the resource manager and the job coordinator — a single process responsible for scheduling tasks across the entire cluster and tracking the state of every running job. This monolithic design meant that a JobTracker failure killed every running job on the cluster. YARN’s split of responsibilities (ResourceManager for cluster-level resource allocation, ApplicationMaster for per-job coordination) is an application of the single-responsibility principle at the distributed systems level. Each component can fail independently, and each has its own recovery mechanism. This architecture mirrors the separation between Kubernetes’ control plane (kube-scheduler, kube-controller-manager) and per-pod lifecycle management. The work-preserving RM restart feature (added in Hadoop 2.6) was a particularly important milestone: it meant that a ResourceManager restart no longer killed running applications, because the RM could reconstruct its state from the running NodeManagers and ApplicationMasters.NodeManager Failure Handling
ResourceManager High Availability
YARN HA Configuration
Building Fault-Tolerant Applications
Designing fault-tolerant applications on Hadoop requires thinking about a property that is deceptively simple to state and surprisingly difficult to achieve in practice: idempotency. An operation is idempotent if executing it multiple times produces the same result as executing it once. This matters in Hadoop because the framework will re-execute tasks on failure, and speculative execution may run duplicate copies of the same task concurrently. If your mapper writes a record to a database on every invocation, a task retry will create duplicate records. This is not a Hadoop-specific concern — it is the central challenge of exactly-once semantics in any distributed system. Kafka addressed it with idempotent producers and transactional writes. Flink uses a two-phase commit protocol with external systems. The patterns shown below (deterministic output paths, upserts instead of inserts, and two-phase commit via OutputCommitter) are foundational techniques that apply far beyond Hadoop.Idempotent Operations
Handling External Systems
Checkpointing Strategies
Monitoring and Health Checks
Health Check Scripts
Monitoring Configuration
Interview Questions
Q1: Explain how HDFS handles DataNode failures. What happens to blocks stored on a failed node?
Q1: Explain how HDFS handles DataNode failures. What happens to blocks stored on a failed node?
Answer:Detection:Key Points:
- NameNode monitors DataNodes via heartbeats (every 3 seconds)
- After 10 missed heartbeats (30 seconds), DataNode marked as stale
- After 10 minutes of no heartbeat, DataNode marked as dead
- Identify affected blocks: NameNode scans metadata to find all blocks that were stored on the failed DataNode
- Determine under-replicated blocks: For each affected block, check if it now has fewer replicas than the configured replication factor
-
Prioritize re-replication:
- Blocks with 0 replicas (highest priority - data loss risk)
- Blocks with 1 replica (high priority)
- Blocks below target replication (normal priority)
-
Schedule re-replication:
- Select source DataNode (has healthy replica, low load)
- Select target DataNode (free space, follows rack-awareness)
- Issue replication command
- Copy blocks: Source DataNode streams block data to target DataNode
- Verify and update: Target verifies checksum, reports to NameNode, metadata updated
- No data loss if at least one replica survives
- Recovery happens automatically
- Client reads are not affected (can use remaining replicas)
- Process is gradual to avoid network saturation
Q2: What is speculative execution in MapReduce? When should you disable it?
Q2: What is speculative execution in MapReduce? When should you disable it?
Answer:Speculative Execution:When to Disable:Best Practice: Enable for pure computation, disable for side effects.
- Mechanism to handle slow tasks (stragglers)
- Framework launches duplicate task on different node
- First task to complete wins, other is killed
- Improves job completion time at cost of extra resources
-
Non-Idempotent Operations:
- Tasks with side effects (database writes, API calls)
- Duplicate execution causes problems
- Example: Incrementing external counter
-
High Resource Utilization:
- Cluster near capacity
- Extra task copies compete for resources
- May actually slow down job
-
Tasks with External Dependencies:
- Rate-limited API calls
- License-limited software
- Shared external resources
-
Known Data Skew:
- Some tasks legitimately take longer (processing more data)
- Speculation wastes resources
- Better to handle via custom partitioning
-
Debugging:
- Investigating task failures
- Want to see actual failure, not masked by successful backup
Q3: How does NameNode High Availability work? Explain the role of QJM and ZooKeeper.
Q3: How does NameNode High Availability work? Explain the role of QJM and ZooKeeper.
Answer:Architecture Components:ZooKeeper’s Role:Failover Process:Why Both QJM and ZooKeeper?:
- Active NameNode: Serves all client requests
- Standby NameNode: Hot standby, ready to take over
- Quorum Journal Manager (QJM): Shared edit log storage
- ZooKeeper: Leader election and fencing
- ZooKeeper Failover Controller (ZKFC): Monitors NameNode health
- QJM: Data plane (edit log storage and replication)
- ZooKeeper: Control plane (coordination, leader election)
- Separation of concerns
- Each specialized for its task
Q4: How do you make MapReduce tasks idempotent? Why is idempotency important for fault tolerance?
Q4: How do you make MapReduce tasks idempotent? Why is idempotency important for fault tolerance?
Answer:Why Idempotency Matters:2. Use Hadoop Counters:3. Deterministic Output Paths:4. Database Operations - Use Unique Keys:5. Write in OutputFormat, Not Map/Reduce:6. Two-Phase Commit:Testing for Idempotency:Key Principle: Design tasks as if they will be executed multiple times, because they might be.
- Tasks may be retried on failure
- Speculative execution runs duplicate tasks
- Non-idempotent operations cause:
- Duplicate records
- Incorrect counters
- Data corruption
- Side effect chaos
Q5: A MapReduce job is running slowly because one reducer is processing 80% of the data (data skew). The ApplicationMaster keeps killing and restarting this reducer. How would you diagnose and fix this?
Q5: A MapReduce job is running slowly because one reducer is processing 80% of the data (data skew). The ApplicationMaster keeps killing and restarting this reducer. How would you diagnose and fix this?
Answer:Diagnosis:1. Identify the Problem:2. Analyze Data Distribution:Root Cause: Data skew - one key has disproportionate amount of dataSolutions:1. Custom Partitioner (Salting):2. Combiner (Reduce Hot Key Volume):3. Increase Reducer Memory/Timeout:4. Use Composite Keys (For Joins):5. Separate Hot Keys:6. Disable Speculative Execution:Prevention:Best Solution: Depends on use case
- Aggregation: Use salting + two-stage aggregation
- Joins: Use map-side join for hot keys
- Counting: Use combiner
- One-off: Increase resources/timeout
Summary
Fault tolerance is the cornerstone of Hadoop’s reliability. The system is designed with the assumption that failures are normal, not exceptional. Through sophisticated mechanisms like data replication, task retries, speculative execution, and high availability architectures, Hadoop achieves remarkable resilience. The patterns established by Hadoop’s fault tolerance design — replication for durability, heartbeat-based failure detection, speculative execution for stragglers, and quorum-based consensus for metadata — have become standard building blocks in modern distributed systems. You will find these same patterns in Apache Kafka (ISR-based replication), Apache Cassandra (tunable replication and hinted handoff), Kubernetes (pod health checks and restart policies), and cloud-native databases like CockroachDB (Raft-based replication with automatic rebalancing). Learning Hadoop’s fault tolerance is not just about Hadoop — it is about learning the vocabulary and design patterns of distributed system reliability that transfer to every system you will build or operate. Key Takeaways:- HDFS Protection: Multi-replica storage, rack awareness, and automatic recovery ensure no data loss
- MapReduce Resilience: Task-level retries, ApplicationMaster recovery, and speculative execution handle compute failures
- YARN Reliability: ResourceManager HA, work-preserving restart, and container recovery maintain cluster availability
- Design Principles: Idempotent operations, checkpointing, and proper external system handling make applications fault-tolerant
- Monitoring: Continuous health checks and proactive monitoring catch issues before they become critical
Interview Deep-Dive
Explain the trade-offs between 3x replication and erasure coding in HDFS. When would you use each, and what happens during a failure recovery in each model?
Explain the trade-offs between 3x replication and erasure coding in HDFS. When would you use each, and what happens during a failure recovery in each model?
Strong Answer:The fundamental trade-off is storage overhead versus recovery performance. 3x replication stores three complete copies of every block, consuming 200% extra storage (3TB stored for every 1TB of data). Erasure coding with the RS-6-3 policy stores 6 data blocks and 3 parity blocks, consuming only 50% extra storage (1.5TB for every 1TB of data). For a petabyte-scale cluster, switching from replication to erasure coding for cold data can save hundreds of terabytes of disk, which translates directly into fewer servers and lower operational cost.The recovery behavior is fundamentally different. When a node fails with 3x replication, recovery is simple: the NameNode picks a healthy DataNode that has a copy of the block and tells it to send that copy to a new target. This is a straightforward block copy — one network transfer per block, no computation required. It is fast and cheap.With erasure coding, recovery requires reconstruction. If one of the 9 blocks (6 data + 3 parity) is lost, the NameNode must coordinate reading at least 6 of the remaining 8 blocks from their respective DataNodes, perform a matrix multiplication in Galois Field arithmetic to reconstruct the missing block, and write the result to a new DataNode. This means recovery reads 6 blocks worth of data over the network (versus 1 block for replication), uses significant CPU for the mathematical reconstruction, and takes longer to complete. During recovery, degraded reads also require this reconstruction, adding latency to client reads.When to use each: Use 3x replication for hot data — data that is actively read by MapReduce jobs, Spark queries, or HBase. The local read optimization (reading from the same node that stores the block) only works with replication, and the fast recovery ensures that failures do not impact running jobs significantly. Use erasure coding for cold data — archived logs, historical data that is rarely accessed, backup copies. The storage savings are substantial, and the occasional higher-latency read is acceptable for data that is accessed infrequently.The practical guideline many organizations use: data less than 30 days old gets 3x replication (hot tier), data between 30 days and 1 year gets RS-6-3 erasure coding (warm tier), and data older than 1 year gets moved to cloud cold storage (S3 Glacier or equivalent).Follow-up: Can you lose data with erasure coding? Under what conditions?With RS-6-3, you can tolerate the simultaneous loss of any 3 out of 9 blocks. If you lose 4 or more blocks simultaneously (before recovery can reconstruct the missing ones), the data is unrecoverable. This is why rack awareness matters even more with erasure coding — the 9 blocks must be spread across enough racks that a single rack failure (which could take out multiple blocks) does not exceed the tolerance threshold. The HDFS erasure coding block placement policy ensures that no two blocks from the same stripe are placed on the same rack. A correlated failure (like a power outage affecting multiple racks) is the primary risk scenario.
The NameNode is Hadoop's single point of failure. Walk me through the HA architecture, explain the role of every component, and describe a failure scenario where the HA mechanism itself fails.
The NameNode is Hadoop's single point of failure. Walk me through the HA architecture, explain the role of every component, and describe a failure scenario where the HA mechanism itself fails.
Strong Answer:NameNode HA uses five components working together. The Active NameNode handles all client requests and writes metadata changes (edit log entries) to the Quorum Journal Manager. The Standby NameNode continuously reads these edit log entries from the QJM and applies them to its own in-memory namespace, keeping itself synchronized. The Quorum Journal Manager is a set of 3 or more JournalNodes that store the edit log — writes must be acknowledged by a majority (quorum) before the Active NameNode considers them durable. The ZooKeeper Failover Controller (ZKFC) runs on each NameNode machine and monitors the local NameNode’s health via RPC calls and heartbeats. The ZooKeeper ensemble provides the leader election mechanism — the ZKFC of the Active NameNode holds an ephemeral znode (lock), and if that ZKFC stops sending heartbeats (because its NameNode died), the lock is released and the Standby’s ZKFC can acquire it.Normal failover sequence: Active NameNode crashes. Its ZKFC detects the failure (health check timeout). ZKFC’s ZooKeeper session expires, releasing the ephemeral lock. Standby’s ZKFC acquires the lock. Standby ZKFC initiates fencing — it attempts to SSH into the old Active’s machine and kill the NameNode process (to prevent a zombie Active from corrupting the edit log). Standby reads any remaining edit log entries from QJM to catch up to the latest state. Standby promotes itself to Active and begins serving client requests. Total failover time: typically 10-30 seconds.Now, the failure scenario where HA itself breaks — the split-brain. Imagine a network partition where the Active NameNode is still running and serving clients on one side of the partition, but the ZKFC’s connection to ZooKeeper is severed. ZooKeeper thinks the Active is dead (session timeout), releases the lock, and the Standby promotes itself. Now you have two Active NameNodes. Both are accepting writes. Both are writing to the QJM. This is catastrophic because the edit log becomes inconsistent.Hadoop prevents this through Epoch Fencing in the QJM. When the new Active NameNode is promoted, it is assigned a monotonically increasing epoch number. It sends this epoch to all JournalNodes. The JournalNodes will reject any write with an epoch lower than the current one. So when the old Active (with the old epoch) tries to write to the QJM, its writes are rejected, it receives an error, and it shuts itself down. This is the critical safety mechanism: the QJM acts as a distributed fence, and the epoch number is the proof of authority. Without the QJM’s epoch fencing, NameNode HA would be fundamentally unsafe.Follow-up: What happens if ZooKeeper itself loses a majority of its nodes?If ZooKeeper loses a majority (e.g., 2 out of 3 ZK nodes), it cannot form a quorum and stops serving requests entirely. This means no failover can happen — the ZKFC cannot acquire or release locks. The Active NameNode continues serving normally (it does not depend on ZooKeeper for ongoing operations, only for failover), but if the Active NameNode fails while ZooKeeper is down, there is no automatic failover. The Standby remains standby. This is why ZooKeeper should run on at least 5 nodes in production (tolerating 2 failures) and should never share machines with other high-resource services that might cause OOM kills.
A MapReduce job writes results to an external database. After a task failure and retry, you discover duplicate records in the database. Explain why this happened and design a solution that guarantees exactly-once semantics.
A MapReduce job writes results to an external database. After a task failure and retry, you discover duplicate records in the database. Explain why this happened and design a solution that guarantees exactly-once semantics.
Strong Answer:This is the classic exactly-once delivery problem in distributed systems, and it happens because MapReduce provides at-least-once execution semantics, not exactly-once. When a map or reduce task fails and is retried, the new attempt re-processes the same input from the beginning. If the failed task had already written some records to the external database before crashing, those records persist in the database. The retry writes them again, creating duplicates.The root cause is that the database write and the task completion are not atomic. The task wrote to the database (side effect committed), then crashed before reporting success to the ApplicationMaster. From MapReduce’s perspective, the task never completed, so it must be retried. From the database’s perspective, the records exist.There are three solution patterns, in increasing order of robustness.Pattern 1 — Idempotent writes with natural keys. If each input record has a natural unique identifier, use UPSERT or REPLACE INTO with that identifier as the primary key. When the retry writes the same records, they overwrite the existing ones rather than creating duplicates. This is the simplest solution and works when your data has natural keys. Example: if processing user events, use
(user_id, event_timestamp) as the composite key.Pattern 2 — OutputCommitter with staging. Do not write to the final database table during the task. Instead, write to a staging table whose name includes the TaskAttemptID. In the OutputCommitter’s commitTask method (which is called exactly once for the successful attempt), atomically move data from the staging table to the final table. In abortTask (called for failed attempts), drop the staging table. This provides atomicity: either all records from a task attempt appear in the final table, or none do.Pattern 3 — Two-phase commit with transaction IDs. For the strongest guarantees, assign a unique transaction ID to each task attempt (using the TaskAttemptID). Write all records within a database transaction tagged with this ID. In the commit phase, mark the transaction as committed. On retry, first check if the previous attempt’s transaction was committed — if so, skip reprocessing. This is essentially implementing a distributed transaction protocol between MapReduce and the database.The honest truth is that true exactly-once semantics across a distributed computation engine and an external system is extremely difficult. Most production systems settle for “effectively once” — idempotent operations that produce the correct final result even if individual operations execute multiple times. This is the approach taken by Kafka’s exactly-once semantics (which is really idempotent production plus transactional consumption) and Flink’s checkpoint-based approach (which provides exactly-once within the Flink pipeline but requires idempotent sinks for external systems).Follow-up: How does Apache Flink solve this problem differently than MapReduce?Flink uses distributed snapshots (the Chandy-Lamport algorithm) to create consistent checkpoints of the entire pipeline state. When a failure occurs, Flink rolls back to the last checkpoint and replays input from that point. Combined with Kafka’s offset management, this means Flink reprocesses only the records since the last checkpoint. For external sinks, Flink provides a two-phase commit sink that writes to a staging area during processing and commits atomically when a checkpoint completes. This provides end-to-end exactly-once semantics without requiring the application developer to implement idempotency manually. The trade-off is that Flink’s checkpointing adds latency (the checkpoint barrier must propagate through the entire DAG) and requires the source to support replay (Kafka does, a bare TCP socket does not).Your 1000-node Hadoop cluster experiences a rack switch failure that takes down an entire rack of 40 nodes simultaneously. Walk through what happens in HDFS, YARN, and any running MapReduce jobs.
Your 1000-node Hadoop cluster experiences a rack switch failure that takes down an entire rack of 40 nodes simultaneously. Walk through what happens in HDFS, YARN, and any running MapReduce jobs.
Strong Answer:This is a correlated failure scenario — the most dangerous type because it violates the assumption that failures are independent. Let me trace through each layer.HDFS response: The NameNode stops receiving heartbeats from all 40 DataNodes. After the heartbeat timeout (default 10 minutes, but the stale interval of 30 seconds triggers earlier marking), it marks all 40 nodes as dead. It then scans its block map to identify every block that had a replica on those nodes. With the default replication factor of 3 and rack-aware placement (2 replicas on one rack, 1 on another), losing an entire rack means every block that had 2 of its 3 replicas on that rack is now down to 1 replica. Blocks that had only 1 replica on the failed rack still have 2 replicas and are less urgent. The NameNode prioritizes re-replication: blocks with 1 remaining replica get highest priority (one more failure means data loss), blocks with 2 remaining replicas get normal priority. The re-replication creates a burst of cross-rack network traffic as the remaining 960 nodes copy blocks to restore the replication factor. On a petabyte-scale cluster, this can take hours and significantly impacts cluster performance during recovery.YARN response: The ResourceManager marks all 40 NodeManagers as lost. All containers running on those nodes — map tasks, reduce tasks, ApplicationMasters — are killed. For each affected application: if the ApplicationMaster was on a failed node, YARN restarts it on a healthy node (up to the maximum AM attempts). If the ApplicationMaster survived but tasks were on failed nodes, the AM marks those tasks as failed and reschedules them on healthy nodes. Reduce tasks that were in the shuffle phase may need to re-fetch map outputs that were on failed nodes — if the map output nodes are among the 40 dead nodes, those map tasks must be re-executed first.MapReduce job impact: Running jobs are disrupted but not killed (assuming the ApplicationMaster survives or is restarted). Map tasks on failed nodes are retried on other nodes. Reduce tasks that had fetched shuffle data from failed nodes must re-fetch from surviving map output locations or wait for map tasks to be re-run. The total job completion time increases significantly — potentially doubling or more depending on how much work was on the failed rack. Speculative execution helps if it was enabled, but 40 nodes lost simultaneously is beyond what speculation is designed to handle.The critical nuance: if the failed rack happened to contain a JournalNode (for NameNode HA), you lose one of your 3 JournalNodes. The QJM still has a quorum (2 of 3) and continues operating, but you have lost fault tolerance for the edit log. Restoring the JournalNode should be a high-priority task. Similarly, if a ZooKeeper node was on the failed rack, the ZK ensemble loses one member and is running degraded.Follow-up: How would you design the rack layout to minimize the blast radius of this type of failure?The key principle is spreading critical services across racks so that no single rack failure takes down a quorum of any service. Place JournalNodes on 3 different racks. Place ZooKeeper nodes on 3 or 5 different racks. Never put both NameNodes on the same rack. Never put both ResourceManagers on the same rack. For data, the rack-aware placement policy already handles this — but verify that it is correctly configured by checking the topology script. Additionally, consider increasing the replication factor for critical datasets to 4 or 5 so that a rack failure leaves more surviving replicas. Finally, implement monitoring that alerts on rack-level health (aggregating DataNode heartbeat status by rack) so that you detect a rack switch degradation before a full failure.
Next Chapter: Chapter 7 - Performance Optimization - Tuning and optimizing Hadoop for maximum performance.