Distributed File Systems

Example: video.mp4 (1 GiB file)

Chunk size: 128 MiB
Result:
  - Chunk 0: bytes 0-134,217,727
  - Chunk 1: bytes 134,217,728-268,435,455
  - Chunk 2: bytes 268,435,456-402,653,183
  - ...
  - Chunk 7: bytes 939,524,096-1,073,741,823

Total: 8 chunks

Chunk 0 replicas:
  - Primary: DataNode 1 (Rack A)
  - Secondary: DataNode 5 (Rack A)
  - Tertiary: DataNode 12 (Rack B)

Purpose:
  - Survive disk failures
  - Survive rack failures
  - Load balancing for reads

# Simplified metadata structure
metadata = {
    "/user/data/video.mp4": {
        "size": 1073741824,  # 1 GiB
        "blocks": [
            {
                "block_id": "blk_1234",
                "size": 134217728,
                "replicas": [
                    {"datanode": "dn1.example.com:50010", "rack": "/rack1"},
                    {"datanode": "dn5.example.com:50010", "rack": "/rack1"},
                    {"datanode": "dn12.example.com:50010", "rack": "/rack2"}
                ]
            },
            # ... 7 more blocks
        ],
        "replication": 3,
        "block_size": 134217728
    }
}

1. Client: "Read /data/video.mp4"
   
2. NameNode responds:
   Block 0 → [DN1, DN5, DN12]
   Block 1 → [DN2, DN6, DN13]
   ...
   
3. Client connects directly to DataNodes:
   - Read Block 0 from DN1 (closest)
   - Read Block 1 from DN2
   - ...
   - Reassemble file
   
4. NameNode NOT involved in data transfer (avoids bottleneck)

class HDFSClient:
    def __init__(self, namenode_url):
        self.namenode = namenode_url
        
    def read_file(self, path):
        # 1. Get block locations from NameNode
        response = requests.post(f"{self.namenode}/getBlockLocations", 
                                json={"path": path})
        blocks = response.json()["blocks"]
        
        # 2. Read each block from DataNodes
        file_data = bytearray()
        
        for block in blocks:
            # Choose closest replica
            datanode = self.choose_datanode(block["replicas"])
            
            # Read block
            block_data = self.read_block(datanode, block["block_id"])
            file_data.extend(block_data)
        
        return bytes(file_data)
    
    def choose_datanode(self, replicas):
        """Choose closest/healthiest replica"""
        # Prefer local DataNode
        for replica in replicas:
            if self.is_local(replica["datanode"]):
                return replica["datanode"]
        
        # Otherwise, choose randomly
        return random.choice(replicas)["datanode"]
    
    def read_block(self, datanode, block_id):
        """Read block from DataNode"""
        response = requests.get(
            f"http://{datanode}/readBlock",
            params={"blockId": block_id}
        )
        return response.content

1. Client: "Write file.txt (300MB)"
   
2. Client asks NameNode: "Where should I write?"
   
3. NameNode responds:
   Block 0 → Write to [DN1, DN2, DN3] (pipeline)
   Block 1 → Write to [DN4, DN5, DN6]
   Block 2 → Write to [DN7, DN8, DN9]
   
4. Client writes Block 0:
   Client → DN1 → DN2 → DN3 (replication pipeline)
   
5. After all blocks written:
   Client → NameNode: "Commit writes"
   NameNode updates metadata

def write_block_with_replication(client_data, datanodes):
    """
    Write block with pipelined replication
    
    Flow:
    Client → DN1 → DN2 → DN3
             ↓     ↓     ↓
           disk  disk  disk
    """
    
    # Client sends to first DataNode
    pipeline = create_pipeline(datanodes)
    
    # DN1 receives, writes to disk, forwards to DN2
    # DN2 receives, writes to disk, forwards to DN3
    # DN3 receives, writes to disk, sends ACK
    
    # ACKs propagate back:
    # DN3 → DN2 → DN1 → Client
    
    send_to_datanode(datanodes[0], client_data, pipeline)
    
    # Wait for ACK from entire pipeline
    return wait_for_ack()

Scenario: DN2 fails (disk crash)

NameNode detects:
  - No heartbeat from DN2 for 10 minutes
  - Marks DN2 as dead
  
Re-replication:
  - Finds under-replicated blocks
  - Block X had replicas on [DN1, DN2, DN5]
  - Now only [DN1, DN5] available
  - Instructs DN1 to replicate to DN8
  - Replication factor restored to 3

Active NameNode ←→ Shared Edit Log ←→ Standby NameNode
                      (JournalNodes store edits; ZooKeeper coordinates automatic failover)

Failover:
1. Active NameNode crashes
2. ZooKeeper detects failure
3. Standby promoted to Active
4. Standby finishes replaying edits before serving as Active
5. Standby becomes new Active
Recovery time depends on detection, fencing, replay, and configuration

class NameNode:
    def checkpoint_metadata(self):
        """Save metadata to disk periodically"""
        fsimage = {
            "namespace": self.namespace_tree,
            "block_map": self.block_locations,
            "replication_queues": self.replication_work
        }
        
        with open("/hdfs/fsimage", "wb") as f:
            pickle.dump(fsimage, f)
        
        # Also save edit log
        self.edit_log.flush()
    
    def recover_from_crash(self):
        """Recover from last checkpoint"""
        # Load fsimage
        with open("/hdfs/fsimage", "rb") as f:
            state = pickle.load(f)
        
        # Replay edit log since checkpoint
        self.replay_edit_log()
        
        # Wait for DataNodes to re-register
        self.wait_for_datanodes()

class TaskScheduler:
    def schedule_map_task(self, input_block):
        """Schedule task on node with data"""
        
        # Get block locations
        replicas = namenode.get_block_locations(input_block)
        
        # Prefer node with data (avoid network transfer)
        for replica in replicas:
            if self.is_worker_available(replica.datanode):
                return self.assign_task(replica.datanode, input_block)
        
        # If data node busy, find nearby node (same rack)
        for replica in replicas:
            nearby_workers = self.get_rack_workers(replica.rack)
            for worker in nearby_workers:
                if self.is_worker_available(worker):
                    return self.assign_task(worker, input_block)
        
        # Last resort: remote read
        return self.assign_task(self.get_any_worker(), input_block)

class HDFSClient:
    def __init__(self):
        self.block_cache = LRUCache(capacity=1000)  # Cache 1000 blocks
    
    def read_block(self, block_id, datanode):
        # Check cache first
        if block_id in self.block_cache:
            return self.block_cache[block_id]
        
        # Read from DataNode
        data = self.fetch_from_datanode(block_id, datanode)
        
        # Cache for future reads
        self.block_cache[block_id] = data
        
        return data