🧠 HDFS Fundamentals
🎯 What is HDFS?
The Hadoop Distributed File System (HDFS) is a distributed, scalable, and fault-tolerant file system designed to store very large files across commodity hardware. It's the storage layer of the Hadoop ecosystem and powers Big Data processing at companies like Yahoo, Facebook, and LinkedIn.
Why HDFS Matters
Scale: Petabytes of data. Fault-tolerant: Handles failures gracefully. Distributed: Data spread across nodes. Cost-effective: Commodity hardware. Throughput: High sequential read/write. Foundation: Powers MapReduce, Spark, Hive.
📌 Distributed📌 Fault-Tolerant📌 Scalable📌 Commodity Hardware
# HDFS — the big picture
# HDFS is Hadoop's distributed file system.
# CONCEPTUAL OVERVIEW
print("""
=== HDFS Overview ===
HDFS is designed for:
1. Very large files (GB to TB per file)
2. Streaming data access (write once, read many)
3. Commodity hardware (failures are normal)
4. High throughput (not low latency)
5. Data locality (compute near data)
Key components:
- NameNode: Master, manages metadata
- DataNode: Worker, stores actual data
- Blocks: Files split into fixed-size blocks
- Replication: Each block stored 3x (default)
HDFS is NOT good for:
- Low-latency access (use HBase/Cassandra)
- Many small files (metadata overhead)
- Multiple writers (append-only)
- Random writes (sequential access)
""")
# HDFS vs TRADITIONAL FILE SYSTEMS
print("=== HDFS vs Traditional FS ===")
print(f"{'Aspect':<25} {'Traditional FS':<30} {'HDFS':<30}")
print("-" * 85)
print(f"{'Storage':<25} {'Single machine':<30} {'Distributed cluster':<30}")
print(f"{'File size':<25} {'Small to medium':<30} {'Very large (GB-TB)':<30}")
print(f"{'Access pattern':<25} {'Random':<30} {'Sequential (streaming)':<30}")
print(f"{'Write pattern':<25} {'Multiple writers':<30} {'Write once, read many':<30}")
print(f"{'Fault tolerance':<25} {'RAID':<30} {'Replication (3x)':<30}")
print(f"{'Hardware':<25} {'Enterprise':<30} {'Commodity':<30}")
print(f"{'Cost':<25} {'High':<30} {'Low':<30}")
print(f"{'Latency':<25} {'Low':<30} {'High (batch)':<30}")
print(f"{'Throughput':<25} {'Medium':<30} {'Very high':<30}")
# KEY PROPERTIES
print("""
=== Key Properties of HDFS ===
1. DISTRIBUTED STORAGE
- Data spread across many machines
- Each machine stores blocks
- Scales horizontally
2. FAULT TOLERANCE
- Replication (default 3x)
- Automatic recovery
- Handles node failures
3. HIGH THROUGHPUT
- Designed for batch processing
- Sequential read/write
- Data locality
4. SCALABILITY
- Add more nodes
- Linear scale
- Petabytes of data
5. SIMPLIFIED CONSISTENCY
- Write once, read many
- No random writes
- Single writer
6. DATA LOCALITY
- Compute moves to data
- Reduces network traffic
- MapReduce/Spark benefit
7. COMMODITY HARDWARE
- No special hardware
- Cost-effective
- Failures expected
""")
# REAL-WORLD SCALE
print("=== Real-World HDFS Deployments ===")
deployments = [
('Yahoo', '40,000 nodes', 'Petabytes'),
('Facebook', '10,000+ nodes', '100+ PB'),
('LinkedIn', '3,000+ nodes', 'Petabytes'),
('Twitter', '10,000+ nodes', 'Petabytes'),
('Netflix', 'Multi-PB', 'Petabytes'),
('Alibaba', '10,000+ nodes', 'Petabytes'),
]
print(f"{'Company':<15} {'Cluster Size':<20} {'Data Volume':<20}")
print("-" * 55)
for company, size, volume in deployments:
print(f"{company:<15} {size:<20} {volume:<20}")
# HDFS USE CASES
print("\n=== HDFS Use Cases ===")
print("""
1. BIG DATA PROCESSING
- MapReduce jobs
- Spark analytics
- Hive queries
- Pig scripts
2. DATA WAREHOUSING
- Store raw data
- ETL pipelines
- Historical archives
3. LOG STORAGE
- Web server logs
- Application logs
- Security logs
4. MACHINE LEARNING
- Training data
- Feature stores
- Model artifacts
5. BACKUP & ARCHIVAL
- Long-term storage
- Compliance data
- Cold data
6. MEDIA STORAGE
- Video files
- Image archives
- Audio recordings
""")
🏗️ Architecture
🎯 Master-Slave Architecture
HDFS follows a master-slave architecture: one NameNode (master) manages metadata, and many DataNodes (slaves) store the actual data. This design separates namespace management from data storage.
Architecture Overview
NameNode: Master, metadata. DataNode: Workers, data blocks. Client: Reads/writes files. Secondary NameNode: Checkpoint helper. Rack: Physical grouping. Heartbeat: Health monitoring.
🏗️ Master-Slave📋 NameNode💾 DataNodes🔄 Heartbeats
# HDFS Architecture
print("=== HDFS Architecture Diagram ===")
print("""
┌─────────────────────┐
│ Client │
│ (HDFS Client) │
└──────────┬──────────┘
│
┌────────────────┼────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ NameNode │ │ Secondary │
│ (Master) │◄────────────►│ NameNode │
│ │ │ (Checkpoint) │
│ - Metadata │ └─────────────────┘
│ - Namespace │
│ - Block map │
└────────┬────────┘
│
│ Heartbeats + Block reports
│
┌────────┴────────┬────────────────┬────────────────┐
│ │ │ │
▼ ▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│DataNode │ │DataNode │ │DataNode │ │DataNode │
│ (1) │ │ (2) │ │ (3) │ │ (4) │
│ │ │ │ │ │ │ │
│ Block A │ │ Block A │ │ Block B │ │ Block B │
│ Block C │ │ Block D │ │ Block C │ │ Block D │
└─────────┘ └─────────┘ └─────────┘ └─────────┘
Rack 1 Rack 1 Rack 2 Rack 2
""")
# COMPONENTS
print("\n=== Core Components ===")
print("""
1. NAMENODE (Master)
- Manages file system namespace
- Stores metadata (inodes, block locations)
- Single point of coordination
- Runs on one machine (or HA pair)
Metadata:
- File hierarchy (directories, files)
- Permissions
- Block locations
- Replication factor
2. DATANODE (Workers)
- Store actual data blocks
- Serve read/write requests
- Report to NameNode
- Many per cluster (100s-1000s)
Responsibilities:
- Block storage
- Block replication
- Heartbeat to NameNode
- Block reports
3. CLIENT
- Interacts with HDFS
- Reads/writes files
- Talks to NameNode for metadata
- Talks to DataNodes for data
4. SECONDARY NAMENODE
- NOT a backup NameNode
- Performs checkpoints
- Merges fsimage + edits
- Reduces NameNode restart time
Note: In HA, Standby NameNode does this.
5. RACK
- Physical grouping of nodes
- Network topology awareness
- Rack-aware replication
6. HEARTBEAT
- DataNode → NameNode every 3 seconds
- Indicates liveness
- Missing 10 heartbeats → dead
""")
# ARCHITECTURE DETAILS
print("\n=== Architecture Details ===")
print("""
Namespace:
- Hierarchical file system
- Directories and files
- Similar to Linux FS
Metadata:
- fsimage: Snapshot of namespace
- edits: Transaction log
- Stored on NameNode local disk
Block Storage:
- Files split into blocks
- Default block size: 128 MB
- Each block replicated 3x
Data Flow:
Write: Client → NameNode (get block locations)
Client → DataNode pipeline
Read: Client → NameNode (get block locations)
Client → DataNode (read block)
Rack Awareness:
- Replica 1: Local rack
- Replica 2: Different rack
- Replica 3: Same rack as replica 2
- Ensures fault tolerance
""")
# IMPLEMENTATION
print("\n=== Simulating HDFS Architecture ===")
class NameNode:
"""Simulated HDFS NameNode."""
def __init__(self):
self.files = {} # path -> list of blocks
self.blocks = {} # block_id -> list of DataNode IDs
self.datanodes = {} # node_id -> DataNode
self.block_size = 128 * 1024 * 1024 # 128 MB
def register_datanode(self, node_id, datanode):
self.datanodes[node_id] = datanode
def create_file(self, path, size):
"""Create file and allocate blocks."""
num_blocks = (size + self.block_size - 1) // self.block_size
block_ids = []
for i in range(num_blocks):
block_id = f"{path}_block_{i}"
block_ids.append(block_id)
# Allocate to datanodes (3x replication)
self.blocks[block_id] = list(self.datanodes.keys())[:3]
self.files[path] = block_ids
return block_ids
def get_block_locations(self, block_id):
return self.blocks.get(block_id, [])
def list_files(self):
return list(self.files.keys())
class DataNode:
"""Simulated HDFS DataNode."""
def __init__(self, node_id):
self.node_id = node_id
self.blocks = {} # block_id -> data
self.used_space = 0
def store_block(self, block_id, data):
self.blocks[block_id] = data
self.used_space += len(data)
def read_block(self, block_id):
return self.blocks.get(block_id)
def delete_block(self, block_id):
if block_id in self.blocks:
self.used_space -= len(self.blocks[block_id])
del self.blocks[block_id]
def list_blocks(self):
return list(self.blocks.keys())
# Simulate cluster
namenode = NameNode()
# Create datanodes
datanodes = {}
for i in range(1, 5):
node_id = f"dn{i}"
dn = DataNode(node_id)
datanodes[node_id] = dn
namenode.register_datanode(node_id, dn)
print(f"Cluster: {len(datanodes)} DataNodes")
# Create a file
file_size = 500 * 1024 * 1024 # 500 MB
blocks = namenode.create_file("/data/logs.txt", file_size)
print(f"\nFile created: /data/logs.txt")
print(f" Size: {file_size / (1024*1024):.0f} MB")
print(f" Blocks: {len(blocks)}")
for b in blocks:
locations = namenode.get_block_locations(b)
print(f" {b} → {locations}")
# Store data in datanodes
for block_id in blocks:
locations = namenode.get_block_locations(block_id)
for dn_id in locations:
datanodes[dn_id].store_block(block_id, f"Data for {block_id}")
print(f"\nDataNode usage:")
for dn_id, dn in datanodes.items():
print(f" {dn_id}: {len(dn.list_blocks())} blocks")
# ARCHITECTURE SUMMARY
print("\n=== Architecture Summary ===")
print(f"{'Component':<25} {'Role':<30} {'Count':<15}")
print("-" * 70)
print(f"{'NameNode':<25} {'Master, metadata':<30} {'1 (or 2 HA)':<15}")
print(f"{'DataNode':<25} {'Store data blocks':<30} {'100s-1000s':<15}")
print(f"{'Secondary NameNode':<25} {'Checkpoint helper':<30} {'1':<15}")
print(f"{'Client':<25} {'Read/write files':<30} {'Many':<15}")
print(f"{'Rack':<25} {'Physical grouping':<30} {'Multiple':<15}")
📋 NameNode
🎯 The Master Node
The NameNode is the master server that manages the file system namespace and regulates access to files. It stores metadata about files, directories, and block locations — but not the actual data.
NameNode Responsibilities
Namespace: File hierarchy. Metadata: File info. Block map: Where blocks live. Permissions: Access control. Edit log: Transaction log. fsimage: Namespace snapshot.
📋 Namespace📋 Metadata📋 Block Map📋 fsimage + edits
# NameNode — the master node
print("=== NameNode Overview ===")
print("""
NameNode is the master of HDFS:
Responsibilities:
1. Manage file system namespace
2. Store metadata (inodes, blocks)
3. Regulate client access
4. Track block locations
5. Handle DataNode heartbeats
6. Manage replication
Does NOT store:
- Actual file data
- Block contents
This separation:
- Allows NameNode to fit in memory
- Fast metadata operations
- Scalable to millions of files
""")
# NAMENODE METADATA
print("\n=== NameNode Metadata ===")
print("""
1. FILE SYSTEM NAMESPACE
- Directory tree
- File names
- File attributes
- Permissions
2. INODES
- File ID
- Size
- Owner, group
- Modification time
- Block list
3. BLOCK MAP
- Block ID → DataNode list
- Replication factor
- Generation stamp
4. TRANSACTION LOG (edits)
- All write operations
- Append-only
- Used for recovery
5. FSIMAGE
- Complete snapshot of namespace
- Point-in-time image
- Merged with edits periodically
""")
# NAMENODE FILES
print("\n=== NameNode Files ===")
print("""
${dfs.namenode.name.dir}/
├── current/
│ ├── VERSION
│ ├── fsimage_0000000000000000000
│ ├── fsimage_0000000000000000000.md5
│ ├── edits_0000000000000000001
│ ├── edits_0000000000000000002
│ ├── ...
│ ├── edits_inprogress_0000000000000000003
│ └── seen_txid
└── in_use.lock
Key files:
- fsimage: Namespace snapshot
- edits: Transaction log
- seen_txid: Last transaction ID
- VERSION: Cluster version info
""")
# NAMENODE MEMORY
print("\n=== NameNode Memory ===")
print("""
NameNode keeps metadata in RAM:
Memory usage:
- ~150 bytes per file object
- ~150 bytes per block
Example:
- 100 million files
- 1 block each
- ~30 GB RAM needed
Rule of thumb:
- 1 GB RAM per 1 million blocks
This limits:
- Number of files
- Number of blocks
- Cluster size
Solutions:
- Federation (multiple NameNodes)
- HDFS-EC (erasure coding)
- Off-heap storage
""")
# NAMENODE OPERATIONS
print("\n=== NameNode Operations ===")
print("""
1. CLIENT REQUESTS
- Create file
- Delete file
- Rename
- List directory
- Get block locations
2. DATANODE MANAGEMENT
- Register DataNode
- Heartbeat (every 3 sec)
- Block report (every 6 hours)
- Decommission
3. BLOCK MANAGEMENT
- Allocate blocks
- Track replicas
- Trigger re-replication
- Balance
4. RECOVERY
- Load fsimage
- Replay edits
- Checkpoint (Secondary NN)
""")
# IMPLEMENTATION
print("\n=== Simulating NameNode ===")
import time
from collections import defaultdict
class NameNodeSimulator:
"""Simulate HDFS NameNode."""
def __init__(self, block_size=128*1024*1024, replication=3):
self.block_size = block_size
self.replication = replication
# Namespace
self.files = {} # path -> FileInfo
self.directories = {'/': []}
# Block map
self.blocks = {} # block_id -> BlockInfo
# DataNodes
self.datanodes = {} # node_id -> DataNodeInfo
# Edit log
self.edits = []
# Statistics
self.total_files = 0
self.total_blocks = 0
def create_file(self, path, size):
"""Create a file."""
num_blocks = (size + self.block_size - 1) // self.block_size
block_ids = []
for i in range(num_blocks):
block_id = f"{path}_blk_{i}"
block_ids.append(block_id)
# Assign to datanodes
locations = list(self.datanodes.keys())[:self.replication]
self.blocks[block_id] = {
'id': block_id,
'size': min(self.block_size, size - i * self.block_size),
'locations': locations,
'generation': 1,
}
self.files[path] = {
'path': path,
'size': size,
'blocks': block_ids,
'replication': self.replication,
'created': time.time(),
}
self.total_files += 1
self.total_blocks += num_blocks
self.edits.append({
'op': 'CREATE',
'path': path,
'time': time.time(),
})
return block_ids
def get_block_locations(self, block_id):
"""Get DataNode locations for a block."""
if block_id in self.blocks:
return self.blocks[block_id]['locations']
return []
def delete_file(self, path):
"""Delete a file."""
if path not in self.files:
return False
file_info = self.files[path]
for block_id in file_info['blocks']:
del self.blocks[block_id]
del self.files[path]
self.total_files -= 1
self.total_blocks -= len(file_info['blocks'])
self.edits.append({
'op': 'DELETE',
'path': path,
'time': time.time(),
})
return True
def register_datanode(self, node_id):
"""Register a DataNode."""
self.datanodes[node_id] = {
'id': node_id,
'last_heartbeat': time.time(),
'blocks': [],
'capacity': 100 * 1024**3, # 100 GB
'used': 0,
}
def heartbeat(self, node_id):
"""Process DataNode heartbeat."""
if node_id in self.datanodes:
self.datanodes[node_id]['last_heartbeat'] = time.time()
def get_stats(self):
"""Get NameNode statistics."""
return {
'total_files': self.total_files,
'total_blocks': self.total_blocks,
'total_datanodes': len(self.datanodes),
'edit_log_size': len(self.edits),
'memory_estimate_mb': (self.total_files * 150 +
self.total_blocks * 150) / (1024*1024),
}
# Simulate
nn = NameNodeSimulator()
# Register DataNodes
for i in range(1, 6):
nn.register_datanode(f"dn{i}")
print(f"Registered {len(nn.datanodes)} DataNodes")
# Create files
files = [
('/data/logs/2024-01-01.log', 500 * 1024**2),
('/data/logs/2024-01-02.log', 750 * 1024**2),
('/data/analytics/sales.parquet', 2 * 1024**3),
('/data/ml/training.csv', 10 * 1024**3),
]
for path, size in files:
blocks = nn.create_file(path, size)
print(f"Created {path} ({size/1024**2:.0f} MB, {len(blocks)} blocks)")
# Stats
stats = nn.get_stats()
print(f"\nNameNode Statistics:")
for key, val in stats.items():
if 'memory' in key:
print(f" {key}: {val:.2f} MB")
else:
print(f" {key}: {val}")
# NAMENODE BOTTLENECK
print("\n=== NameNode Bottleneck ===")
print("""
Problem:
- Single NameNode = single point of failure
- Memory limits cluster size
- Throughput limits
Solutions:
1. High Availability (HA)
- Active + Standby NameNodes
- Automatic failover
- Shared edit log (QJM)
2. Federation
- Multiple NameNodes
- Each manages namespace
- Block pool IDs
3. HDFS Erasure Coding
- Reduces storage overhead
- Less replication needed
4. Off-heap Storage
- Store metadata off-heap
- Larger clusters
""")
# NAMENODE SUMMARY
print("\n=== NameNode Summary ===")
print(f"{'Aspect':<25} {'Description':<50}")
print("-" * 80)
print(f"{'Role':<25} {'Master, metadata manager':<50}")
print(f"{'Storage':<25} {'Metadata only (not data)':<50}")
print(f"{'Memory':<25} {'~150 bytes per file/block':<50}")
print(f"{'Files':<25} {'fsimage, edits':<50}")
print(f"{'Operations':<25} {'Create, delete, read, write':<50}")
print(f"{'Bottleneck':<25} {'Single point, memory':<50}")
print(f"{'Solution':<25} {'HA, Federation':<50}")
💾 DataNode
🎯 The Worker Nodes
DataNodes are the worker nodes that store the actual data blocks. They serve read/write requests from clients and report to the NameNode periodically. Each cluster has many DataNodes (100s to 1000s).
DataNode Responsibilities
Storage: Store data blocks. Read: Serve read requests. Write: Handle write pipeline. Replication: Copy blocks. Heartbeat: Report to NameNode. Block report: List blocks.
💾 Block Storage📖 Read Service✍️ Write Service💓 Heartbeat
# DataNode — the worker nodes
print("=== DataNode Overview ===")
print("""
DataNode is the worker of HDFS:
Responsibilities:
1. Store data blocks
2. Serve read requests
3. Handle write pipeline
4. Replicate blocks
5. Send heartbeats
6. Send block reports
7. Perform block scanning
Each DataNode:
- Stores multiple blocks
- Runs on commodity hardware
- Has local disk storage
- Communicates with NameNode
""")
# DATANODE STORAGE
print("\n=== DataNode Storage ===")
print("""
DataNode stores blocks in local file system:
${dfs.datanode.data.dir}/
├── current/
│ ├── BP-xxxxx-xxx.xxx.xxx.xxx-xxxxxxx/
│ │ ├── current/
│ │ │ ├── finalized/
│ │ │ │ ├── blk_1073741825
│ │ │ │ ├── blk_1073741825_1001.meta
│ │ │ │ ├── blk_1073741826
│ │ │ │ ├── blk_1073741826_1002.meta
│ │ │ │ └── ...
│ │ │ ├── rbw/ (being written)
│ │ │ └── tmp/
│ │ └── VERSION
└── in_use.lock
Block files:
- blk_: actual data
- blk__.meta: metadata (checksum)
Block pool:
- Each namespace has a block pool
- Federation uses multiple block pools
""")
# DATANODE OPERATIONS
print("\n=== DataNode Operations ===")
print("""
1. BLOCK STORAGE
- Receive blocks from clients
- Write to local disk
- Compute checksums
- Report to NameNode
2. BLOCK SERVING
- Receive read requests
- Read from local disk
- Send to client
- Verify checksums
3. REPLICATION
- Receive replication requests
- Copy block from another DataNode
- Report completion
4. HEARTBEAT
- Send every 3 seconds
- Include: capacity, usage, load
- NameNode responds
5. BLOCK REPORT
- Send every 6 hours
- List all blocks
- NameNode reconciles
6. BLOCK SCANNER
- Periodic verification
- Detect corruption
- Report to NameNode
7. DECOMMISSION
- Graceful removal
- Replicate blocks elsewhere
- Then shut down
""")
# HEARTBEAT MECHANISM
print("\n=== Heartbeat Mechanism ===")
print("""
DataNode → NameNode every 3 seconds:
Heartbeat contains:
- DataNode ID
- Capacity
- Used space
- Number of blocks
- Load metrics
NameNode response:
- Acknowledgment
- Commands (replicate, delete)
- Registration info
Failure detection:
- Missing 10 heartbeats (30 sec) → stale
- Missing 30 minutes → dead
- NameNode triggers re-replication
Configuration:
dfs.heartbeat.interval = 3 (seconds)
dfs.namenode.heartbeat.recheck-interval = 300000 (ms)
""")
# BLOCK REPORT
print("\n=== Block Report ===")
print("""
DataNode → NameNode every 6 hours:
Block report contains:
- List of all block IDs
- Generation stamps
- Block lengths
Purpose:
- NameNode reconciles block map
- Detect missing blocks
- Detect extra blocks
- Trigger replication
Types:
- Full block report (all blocks)
- Incremental block report (changes)
Configuration:
dfs.blockreport.intervalMsec = 21600000 (6 hours)
dfs.blockreport.initialDelay = 0
""")
# IMPLEMENTATION
print("\n=== Simulating DataNode ===")
import hashlib
import os
class DataNodeSimulator:
"""Simulate HDFS DataNode."""
def __init__(self, node_id, capacity_gb=100):
self.node_id = node_id
self.capacity = capacity_gb * 1024**3
self.used = 0
# Block storage
self.blocks = {} # block_id -> {data, checksum, generation}
# Stats
self.blocks_read = 0
self.blocks_written = 0
self.last_heartbeat = time.time()
self.last_block_report = time.time()
def store_block(self, block_id, data, generation=1):
"""Store a block."""
checksum = hashlib.md5(data.encode()).hexdigest()
self.blocks[block_id] = {
'data': data,
'checksum': checksum,
'generation': generation,
'size': len(data),
}
self.used += len(data)
self.blocks_written += 1
return checksum
def read_block(self, block_id):
"""Read a block."""
if block_id not in self.blocks:
return None
block = self.blocks[block_id]
self.blocks_read += 1
# Verify checksum
checksum = hashlib.md5(block['data'].encode()).hexdigest()
if checksum != block['checksum']:
raise ValueError(f"Checksum mismatch for {block_id}")
return block['data']
def delete_block(self, block_id):
"""Delete a block."""
if block_id in self.blocks:
self.used -= self.blocks[block_id]['size']
del self.blocks[block_id]
return True
return False
def heartbeat(self):
"""Send heartbeat to NameNode."""
self.last_heartbeat = time.time()
return {
'node_id': self.node_id,
'capacity': self.capacity,
'used': self.used,
'available': self.capacity - self.used,
'num_blocks': len(self.blocks),
'load': self.blocks_read + self.blocks_written,
}
def block_report(self):
"""Send block report to NameNode."""
self.last_block_report = time.time()
return {
'node_id': self.node_id,
'blocks': list(self.blocks.keys()),
'count': len(self.blocks),
}
def scan_blocks(self):
"""Scan blocks for corruption."""
corrupt = []
for block_id, block in self.blocks.items():
checksum = hashlib.md5(block['data'].encode()).hexdigest()
if checksum != block['checksum']:
corrupt.append(block_id)
return corrupt
def get_stats(self):
return {
'node_id': self.node_id,
'capacity_gb': self.capacity / 1024**3,
'used_gb': self.used / 1024**3,
'utilization': self.used / self.capacity,
'num_blocks': len(self.blocks),
'blocks_read': self.blocks_read,
'blocks_written': self.blocks_written,
}
# Simulate
dn = DataNodeSimulator("dn1", capacity_gb=100)
# Store blocks
for i in range(10):
block_id = f"blk_{1000 + i}"
data = f"Data for block {block_id}" * 1000
dn.store_block(block_id, data)
# Stats
stats = dn.get_stats()
print(f"DataNode {stats['node_id']} Statistics:")
print(f" Capacity: {stats['capacity_gb']:.1f} GB")
print(f" Used: {stats['used_gb']:.4f} GB")
print(f" Utilization: {stats['utilization']*100:.2f}%")
print(f" Blocks: {stats['num_blocks']}")
print(f" Blocks written: {stats['blocks_written']}")
# Read a block
data = dn.read_block("blk_1000")
print(f"\nRead block: {len(data)} bytes")
# Heartbeat
hb = dn.heartbeat()
print(f"\nHeartbeat: {hb['num_blocks']} blocks, {hb['used']} bytes used")
# Scan for corruption
corrupt = dn.scan_blocks()
print(f"Corrupt blocks: {len(corrupt)}")
# DATANODE FAILURE
print("\n=== DataNode Failure Handling ===")
print("""
When a DataNode fails:
1. DETECTION
- Missing heartbeats (30 sec)
- NameNode marks as stale
- After 10.5 min → dead
2. RE-REPLICATION
- NameNode identifies under-replicated blocks
- Schedules replication to other nodes
- Prioritizes by replication factor
3. CLIENT IMPACT
- Reads: Client uses other replicas
- Writes: Pipeline recovery
- No data loss (if replicas exist)
4. RECOVERY
- If DataNode returns, it re-registers
- NameNode reconciles blocks
- Deletes extra blocks
5. DECOMMISSION
- Graceful removal
- Replicate blocks first
- Then shut down
""")
# DATANODE SUMMARY
print("\n=== DataNode Summary ===")
print(f"{'Aspect':<25} {'Description':<50}")
print("-" * 80)
print(f"{'Role':<25} {'Worker, stores data':<50}")
print(f"{'Storage':<25} {'Local disk (block files)':<50}")
print(f"{'Operations':<25} {'Read, write, replicate':<50}")
print(f"{'Heartbeat':<25} {'Every 3 seconds':<50}")
print(f"{'Block report':<25} {'Every 6 hours':<50}")
print(f"{'Failure':<25} {'Re-replicate blocks':<50}")
print(f"{'Count':<25} {'100s to 1000s per cluster':<50}")
🧱 Blocks & Replication
🎯 The Building Blocks of HDFS
HDFS splits files into fixed-size blocks (default 128 MB) and replicates each block across multiple DataNodes (default 3x). This provides fault tolerance and enables parallel processing.
Block & Replication
Block size: 128 MB default. Replication: 3x default. Placement: Rack-aware. Fault tolerance: Replicas. Parallelism: One block per mapper. Storage: 3x overhead.
🧱 128 MB Blocks3️⃣ 3x Replication🏢 Rack-Aware🛡️ Fault Tolerant
# HDFS Blocks and Replication
print("=== Block Size ===")
print("""
Default block size: 128 MB (Hadoop 2.x+)
Previous default: 64 MB (Hadoop 1.x)
Why 128 MB?
- Balance between:
- Metadata overhead (too many small blocks)
- Parallelism (too few large blocks)
- Modern hardware: higher throughput
- Larger blocks = less seek overhead
Configurable:
dfs.blocksize = 134217728 (128 MB)
Per-file override:
hdfs dfs -D dfs.blocksize=268435456 -put file.txt /data/
Block size considerations:
- Larger blocks: fewer blocks, less metadata, more sequential
- Smaller blocks: more blocks, more parallelism, more overhead
""")
# BLOCK EXAMPLE
print("\n=== Block Example ===")
print("""
File: 500 MB
Block size: 128 MB
Replication: 3
Blocks:
Block 0: 0-128 MB
Block 1: 128-256 MB
Block 2: 256-384 MB
Block 3: 384-500 MB (116 MB, last block smaller)
Total blocks: 4
Total replicas: 4 × 3 = 12 block replicas
Storage overhead: 3x (500 MB × 3 = 1.5 GB)
""")
# REPLICATION
print("\n=== Replication ===")
print("""
Default replication factor: 3
Why 3?
- Fault tolerance: 2 nodes can fail
- Availability: Always a replica
- Performance: Local reads
Configurable:
dfs.replication = 3
Per-file:
hdfs dfs -setrep -w 5 /data/file.txt
Replication factors:
1: No fault tolerance (test only)
2: 1 node can fail
3: 2 nodes can fail (default)
4+: Higher durability
Storage overhead:
- 3x replication = 3x storage
- Erasure coding: ~1.4x (newer)
""")
# RACK AWARENESS
print("\n=== Rack-Aware Replication ===")
print("""
HDFS is rack-aware:
Default placement (replication=3):
1. Replica 1: Local rack (same as writer)
2. Replica 2: Different rack
3. Replica 3: Same rack as replica 2
Why?
- Fault tolerance: Rack failure
- Performance: Local reads
- Network: Reduce cross-rack traffic
Rack failure:
- 3 replicas in 2 racks
- One rack can fail
- Data still available
Configuration:
net.topology.script.file.name = /etc/hadoop/rack.sh
""")
# BLOCK PLACEMENT
print("\n=== Block Placement ===")
print("""
Writer is on DataNode 1 (Rack 1):
Placement:
Replica 1: DataNode 1 (Rack 1) — local
Replica 2: DataNode 3 (Rack 2) — remote
Replica 3: DataNode 4 (Rack 2) — same rack as replica 2
Result:
Rack 1: 1 replica
Rack 2: 2 replicas
If Rack 1 fails:
- Replicas in Rack 2 available
- Data accessible
If DataNode 1 fails:
- Replicas 2 and 3 available
- Data accessible
""")
# REPLICATION MONITORING
print("\n=== Replication Monitoring ===")
print("""
NameNode tracks:
- Under-replicated blocks
- Over-replicated blocks
- Corrupt blocks
Commands:
hdfs fsck /path -files -blocks -locations
Output:
/data/file.txt 500 MB, 4 blocks
Block 0: 128 MB, 3 replicas
...
Status: HEALTHY
hdfs dfsadmin -report
Output:
Live datanodes: 5
Total capacity: 500 GB
Used: 200 GB
Under-replicated blocks: 0
""")
# IMPLEMENTATION
print("\n=== Simulating Blocks & Replication ===")
class BlockManager:
"""Simulate HDFS block management."""
def __init__(self, block_size=128*1024*1024, replication=3):
self.block_size = block_size
self.replication = replication
self.blocks = {} # block_id -> BlockInfo
self.datanodes = {} # node_id -> DataNode
self.racks = {} # rack_id -> [node_ids]
def add_datanode(self, node_id, rack_id):
"""Add a DataNode to a rack."""
self.datanodes[node_id] = {
'id': node_id,
'rack': rack_id,
'blocks': [],
'capacity': 100 * 1024**3,
'used': 0,
}
if rack_id not in self.racks:
self.racks[rack_id] = []
self.racks[rack_id].append(node_id)
def create_file(self, path, size):
"""Create file and allocate blocks."""
num_blocks = (size + self.block_size - 1) // self.block_size
block_ids = []
for i in range(num_blocks):
block_id = f"{path}_blk_{i}"
block_size = min(self.block_size, size - i * self.block_size)
# Rack-aware placement
locations = self._place_block(block_id)
self.blocks[block_id] = {
'id': block_id,
'size': block_size,
'locations': locations,
'replication': self.replication,
}
block_ids.append(block_id)
return block_ids
def _place_block(self, block_id):
"""Rack-aware block placement."""
# Get all racks
rack_ids = list(self.racks.keys())
if len(rack_ids) < 2:
# Not enough racks — use all nodes
return list(self.datanodes.keys())[:self.replication]
# Replica 1: First rack
rack1 = rack_ids[0]
nodes_rack1 = self.racks[rack1]
# Replica 2 & 3: Second rack
rack2 = rack_ids[1]
nodes_rack2 = self.racks[rack2]
locations = []
# Replica 1
if nodes_rack1:
locations.append(nodes_rack1[0])
# Replica 2
if nodes_rack2:
locations.append(nodes_rack2[0])
# Replica 3
if len(nodes_rack2) > 1:
locations.append(nodes_rack2[1])
elif len(nodes_rack1) > 1:
locations.append(nodes_rack1[1])
return locations[:self.replication]
def get_block_locations(self, block_id):
if block_id in self.blocks:
return self.blocks[block_id]['locations']
return []
def get_stats(self):
under_replicated = 0
over_replicated = 0
for block in self.blocks.values():
replicas = len(block['locations'])
if replicas < self.replication:
under_replicated += 1
elif replicas > self.replication:
over_replicated += 1
return {
'total_blocks': len(self.blocks),
'total_replicas': sum(len(b['locations']) for b in self.blocks.values()),
'under_replicated': under_replicated,
'over_replicated': over_replicated,
'rack_count': len(self.racks),
'datanode_count': len(self.datanodes),
}
# Simulate
bm = BlockManager(block_size=128*1024*1024, replication=3)
# Add DataNodes in 2 racks
for i in range(1, 4):
bm.add_datanode(f"dn{i}", "rack1")
for i in range(4, 7):
bm.add_datanode(f"dn{i}", "rack2")
print(f"Cluster: {len(bm.datanodes)} DataNodes in {len(bm.racks)} racks")
print(f"Rack 1: {bm.racks['rack1']}")
print(f"Rack 2: {bm.racks['rack2']}")
# Create files
files = [
('/data/file1.txt', 500 * 1024**2),
('/data/file2.txt', 1 * 1024**3),
]
for path, size in files:
blocks = bm.create_file(path, size)
print(f"\nCreated {path} ({size/1024**2:.0f} MB, {len(blocks)} blocks)")
for block_id in blocks:
locations = bm.get_block_locations(block_id)
racks = [bm.datanodes[loc]['rack'] for loc in locations]
print(f" {block_id}: {locations} (racks: {racks})")
# Stats
stats = bm.get_stats()
print(f"\nBlock Manager Statistics:")
for key, val in stats.items():
print(f" {key}: {val}")
# REPLICATION SUMMARY
print("\n=== Replication Summary ===")
print(f"{'Aspect':<25} {'Default':<15} {'Configurable':<20}")
print("-" * 60)
print(f"{'Block size':<25} {'128 MB':<15} {'dfs.blocksize':<20}")
print(f"{'Replication':<25} {'3':<15} {'dfs.replication':<20}")
print(f"{'Rack awareness':<25} {'Yes':<15} {'topology script':<20}")
print(f"{'Storage overhead':<25} {'3x':<15} {'Erasure coding':<20}")
📖 Read Path
🎯 How HDFS Reads Data
The read path describes how a client reads a file from HDFS: first contacting the NameNode for block locations, then reading blocks directly from the nearest DataNode.
Read Path Steps
1. Open: Client → NameNode. 2. Locations: NameNode returns block locations. 3. Read: Client → DataNode. 4. Verify: Checksum validation. 5. Next block: Continue until done. 6. Close: Release resources.
📖 Open📍 Get Locations📥 Read Blocks✅ Verify
# HDFS Read Path
print("=== Read Path Overview ===")
print("""
Client wants to read /data/file.txt:
1. OPEN
Client → NameNode: open("/data/file.txt")
2. GET BLOCK LOCATIONS
NameNode → Client: [Block 0: DN1, DN2, DN3], [Block 1: DN2, DN4, DN5], ...
3. READ BLOCK
Client → Nearest DataNode: readBlock(Block 0)
DataNode → Client: block data
4. VERIFY
Client verifies checksum
5. NEXT BLOCK
Repeat for all blocks
6. CLOSE
Client → NameNode: close()
""")
# READ PATH DIAGRAM
print("\n=== Read Path Diagram ===")
print("""
┌─────────┐
│ Client │
└────┬────┘
│ 1. open("/data/file.txt")
▼
┌─────────────┐
│ NameNode │
│ │
│ Metadata: │
│ - Block 0 │
│ - Block 1 │
│ - Block 2 │
└──────┬──────┘
│ 2. Block locations
│ Block 0: [DN1, DN2, DN3]
│ Block 1: [DN2, DN4, DN5]
│ Block 2: [DN1, DN3, DN5]
▼
┌─────────────────────────────────────────┐
│ Client │
│ │
│ 3. Read Block 0 from DN1 (nearest) │
│ 4. Read Block 1 from DN2 │
│ 5. Read Block 2 from DN1 │
└─────────────────────────────────────────┘
│
│ 3-5. Read blocks
▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ DN1 │ │ DN2 │ │ DN3 │
│ │ │ │ │ │
│ Block 0 │ │ Block 0 │ │ Block 0 │
│ Block 2 │ │ Block 1 │ │ Block 2 │
└─────────┘ └─────────┘ └─────────┘
""")
# READ PATH DETAILS
print("\n=== Read Path Details ===")
print("""
1. OPEN
- Client calls FileSystem.open()
- Returns FSDataInputStream
- Client → NameNode RPC
2. GET BLOCK LOCATIONS
- Client calls getBlockLocations()
- NameNode returns:
- Block IDs
- DataNode locations (sorted by network distance)
- Block sizes
Optimization:
- NameNode returns all blocks for file
- Client caches locations
3. READ BLOCK
- Client picks nearest DataNode
- Opens socket to DataNode
- Sends block ID + offset
- DataNode streams data
DataNode selection:
- Local node (if available)
- Same rack
- Different rack
4. VERIFY
- Client receives data + checksum
- Verifies checksum
- If corrupt: try next replica
- Report corrupt block to NameNode
5. NEXT BLOCK
- Client reads next block
- Uses cached locations
- If location unknown: contact NameNode
6. CLOSE
- Client closes stream
- Releases resources
- No NameNode notification needed
""")
# READ OPTIMIZATIONS
print("\n=== Read Optimizations ===")
print("""
1. SHORT-CIRCUIT LOCAL READS
- If client on same node as block
- Bypass DataNode
- Read directly from local disk
Config: dfs.client.read.shortcircuit = true
2. HEDGED READS
- Send request to multiple replicas
- Use first response
- Reduces tail latency
Config: dfs.client.hedged.read.threadpool.size
3. CACHING
- HDFS cache (off-heap)
- Client cache
- OS page cache
Command: hdfs cacheadmin -addPool
4. ZERO-COPY READS
- Send file descriptor
- Avoid copying
- Used for local reads
Config: dfs.datanode.transferTo.allowed
5. PARALLEL READS
- Read multiple blocks
- Used by MapReduce/Spark
- One mapper per block
""")
# IMPLEMENTATION
print("\n=== Simulating Read Path ===")
import random
class HDFSClient:
"""Simulate HDFS client read."""
def __init__(self, namenode, datanodes):
self.namenode = namenode
self.datanodes = datanodes
self.read_stats = {
'blocks_read': 0,
'bytes_read': 0,
'checksum_errors': 0,
}
def read_file(self, path):
"""Read a file from HDFS."""
print(f"Reading {path}...")
# Step 1: Open (get file info from NameNode)
if path not in self.namenode.files:
print(f" File not found: {path}")
return None
file_info = self.namenode.files[path]
print(f" File size: {file_info['size'] / 1024**2:.0f} MB")
print(f" Blocks: {len(file_info['blocks'])}")
# Step 2: Get block locations
data_parts = []
for block_id in file_info['blocks']:
locations = self.namenode.get_block_locations(block_id)
print(f"\n Block {block_id}:")
print(f" Locations: {locations}")
# Step 3: Read from nearest DataNode
block_data = self._read_block(block_id, locations)
if block_data:
data_parts.append(block_data)
self.read_stats['blocks_read'] += 1
self.read_stats['bytes_read'] += len(block_data)
# Step 4: Verify and return
return ''.join(data_parts)
def _read_block(self, block_id, locations):
"""Read block from nearest DataNode."""
# Try each location
for node_id in locations:
if node_id in self.datanodes:
dn = self.datanodes[node_id]
data = dn.read_block(block_id)
if data:
print(f" Read from {node_id}: {len(data)} bytes")
return data
else:
print(f" {node_id}: block not found")
print(f" ERROR: No replica available")
return None
def get_stats(self):
return self.read_stats
# Use the NameNode and DataNodes from previous examples
# (In practice, these would be the same objects)
class SimpleNameNode:
def __init__(self):
self.files = {}
self.blocks = {}
self.datanodes = {}
def register_datanode(self, node_id, dn):
self.datanodes[node_id] = dn
def create_file(self, path, blocks):
self.files[path] = {'size': sum(b['size'] for b in blocks), 'blocks': [b['id'] for b in blocks]}
for b in blocks:
self.blocks[b['id']] = b
def get_block_locations(self, block_id):
if block_id in self.blocks:
return self.blocks[block_id]['locations']
return []
# Setup
nn = SimpleNameNode()
datanodes = {}
# Create DataNodes
for i in range(1, 4):
dn = DataNodeSimulator(f"dn{i}")
datanodes[f"dn{i}"] = dn
nn.register_datanode(f"dn{i}", dn)
# Create a file with 3 blocks
blocks = []
for i in range(3):
block_id = f"blk_{i}"
locations = [f"dn{(i % 3) + 1}", f"dn{((i + 1) % 3) + 1}"]
blocks.append({'id': block_id, 'size': 128 * 1024**2, 'locations': locations})
nn.create_file("/data/test.txt", blocks)
# Store data in DataNodes
for block in blocks:
for node_id in block['locations']:
datanodes[node_id].store_block(block['id'], f"Data for {block['id']}" * 1000)
# Read file
client = HDFSClient(nn, datanodes)
data = client.read_file("/data/test.txt")
print(f"\nRead completed:")
print(f" Total bytes: {client.read_stats['bytes_read']}")
print(f" Blocks read: {client.read_stats['blocks_read']}")
# READ PATH SUMMARY
print("\n=== Read Path Summary ===")
print(f"{'Step':<25} {'Action':<40} {'Component':<20}")
print("-" * 85)
print(f"{'1. Open':<25} {'Client → NameNode':<40} {'NameNode':<20}")
print(f"{'2. Get locations':<25} {'NameNode → Client':<40} {'NameNode':<20}")
print(f"{'3. Read block':<25} {'Client → DataNode':<40} {'DataNode':<20}")
print(f"{'4. Verify':<25} {'Client checksum':<40} {'Client':<20}")
print(f"{'5. Next block':<25} {'Repeat 2-4':<40} {'Client':<20}")
print(f"{'6. Close':<25} {'Release resources':<40} {'Client':<20}")
✍️ Write Path
🎯 How HDFS Writes Data
The write path describes how a client writes a file to HDFS: creating the file in the namespace, then writing blocks through a pipeline of DataNodes with replication.
Write Path Steps
1. Create: Client → NameNode. 2. Pipeline: NameNode → DataNodes. 3. Write: Client → DN1 → DN2 → DN3. 4. ACK: DN3 → DN2 → DN1 → Client. 5. Close: Finalize file. 6. Complete: NameNode commit.
✍️ Create🔗 Pipeline📝 Write✅ ACK
# HDFS Write Path
print("=== Write Path Overview ===")
print("""
Client wants to write /data/newfile.txt:
1. CREATE
Client → NameNode: create("/data/newfile.txt")
NameNode: creates file entry, returns FSDataOutputStream
2. WRITE BLOCK
Client buffers data until block full (128 MB)
3. GET PIPELINE
Client → NameNode: addBlock()
NameNode: returns pipeline [DN1, DN2, DN3]
4. SETUP PIPELINE
Client → DN1 → DN2 → DN3
ACK: DN3 → DN2 → DN1 → Client
5. STREAM DATA
Client → DN1 → DN2 → DN3 (packets)
ACK: DN3 → DN2 → DN1 → Client
6. CLOSE
Client → NameNode: complete()
NameNode: commits file
""")
# WRITE PATH DIAGRAM
print("\n=== Write Path Diagram ===")
print("""
┌─────────┐
│ Client │
└────┬────┘
│ 1. create("/data/newfile.txt")
▼
┌─────────────┐
│ NameNode │
│ │
│ - Create │
│ - Add block│
│ - Complete │
└──────┬──────┘
│ 2. Pipeline: [DN1, DN2, DN3]
▼
┌─────────────────────────────────────────────────┐
│ Client │
│ │
│ 3. Write packet → DN1 │
│ 4. DN1 → DN2 │
│ 5. DN2 → DN3 │
│ 6. ACK: DN3 → DN2 → DN1 → Client │
└─────────────────────────────────────────────────┘
│
▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ DN1 │───►│ DN2 │───►│ DN3 │
│ │ │ │ │ │
│ Block 0 │ │ Block 0 │ │ Block 0 │
│ (replica)│ │ (replica)│ │ (replica)│
└─────────┘ └─────────┘ └─────────┘
Replica 1 Replica 2 Replica 3
""")
# WRITE PATH DETAILS
print("\n=== Write Path Details ===")
print("""
1. CREATE
- Client calls FileSystem.create()
- NameNode:
- Checks permissions
- Checks if file exists
- Creates file entry (no blocks yet)
- Returns FSDataOutputStream
2. WRITE DATA
- Client writes data to stream
- Data buffered in client memory
- When buffer reaches block size:
- Client requests new block
3. ADD BLOCK
- Client → NameNode: addBlock()
- NameNode:
- Allocates new block ID
- Selects pipeline DataNodes
- Returns pipeline + block ID
4. PIPELINE SETUP
- Client → DN1: writeBlock(block, pipeline)
- DN1 → DN2: writeBlock(block, pipeline)
- DN2 → DN3: writeBlock(block, pipeline)
- ACK: DN3 → DN2 → DN1 → Client
5. DATA STREAMING
- Client sends packets (64 KB)
- DN1 forwards to DN2
- DN2 forwards to DN3
- ACKs flow back
- Packet includes checksum
6. CLOSE
- Client closes stream
- Waits for all ACKs
- Client → NameNode: complete()
- NameNode commits file
7. ERROR HANDLING
- If DN fails during write:
- Pipeline recovery
- Replace failed node
- Continue writing
""")
# PIPELINE RECOVERY
print("\n=== Pipeline Recovery ===")
print("""
If a DataNode fails during write:
1. DETECTION
- No ACK from failed node
- Timeout
2. RECOVERY
- Client closes pipeline
- Contacts NameNode
- Gets new pipeline (excluding failed node)
- Continues writing
3. DATA CONSISTENCY
- Blocks already written are valid
- New pipeline replicates to new node
- NameNode updates block map
Example:
Original pipeline: [DN1, DN2, DN3]
DN2 fails
New pipeline: [DN1, DN4, DN3]
DN4 gets replica from DN1
4. ACK HANDLING
- ACKs from remaining nodes
- Failed node removed
- Client continues
""")
# WRITE OPTIMIZATIONS
print("\n=== Write Optimizations ===")
print("""
1. PIPELINE
- Data streamed through pipeline
- Not wait for full block
- Efficient use of network
2. PACKET SIZE
- 64 KB packets
- Balance latency/throughput
- Configurable
3. CHECKSUM
- CRC32 checksum
- Per 512 bytes
- Detect corruption
4. BUFFERING
- Client buffers data
- Reduces RPC calls
- Improves throughput
5. PIPELINE RECOVERY
- Handle failures
- Continue writing
- No data loss
6. LEASE
- Single writer
- Lease renewal
- Prevents conflicts
7. HFLUSH
- Force data to DataNodes
- Visible to readers
- hflush() vs hsync()
""")
# IMPLEMENTATION
print("\n=== Simulating Write Path ===")
class HDFSWriteClient:
"""Simulate HDFS client write."""
def __init__(self, namenode, datanodes):
self.namenode = namenode
self.datanodes = datanodes
self.write_stats = {
'blocks_written': 0,
'bytes_written': 0,
'pipeline_recoveries': 0,
}
def create_file(self, path, data):
"""Write a file to HDFS."""
print(f"Writing {path} ({len(data)} bytes)...")
# Step 1: Create file in NameNode
print(f" 1. Create: {path}")
self.namenode.files[path] = {'size': 0, 'blocks': []}
# Step 2: Split into blocks
block_size = 128 * 1024 # Small for demo
blocks = []
for i in range(0, len(data), block_size):
block_data = data[i:i+block_size]
block_id = f"{path}_blk_{len(blocks)}"
# Step 3: Get pipeline
pipeline = self._get_pipeline()
print(f" 2. Block {block_id}: pipeline {pipeline}")
# Step 4: Write to pipeline
self._write_block(block_id, block_data, pipeline)
blocks.append({
'id': block_id,
'size': len(block_data),
'locations': pipeline,
})
# Step 5: Complete
self.namenode.files[path] = {
'size': len(data),
'blocks': [b['id'] for b in blocks],
}
for b in blocks:
self.namenode.blocks[b['id']] = b
print(f" 3. Complete: {len(blocks)} blocks written")
return blocks
def _get_pipeline(self):
"""Get write pipeline from NameNode."""
# Simple: first 3 DataNodes
return list(self.datanodes.keys())[:3]
def _write_block(self, block_id, data, pipeline):
"""Write block through pipeline."""
# Simulate pipeline write
for node_id in pipeline:
if node_id in self.datanodes:
self.datanodes[node_id].store_block(block_id, data)
self.write_stats['blocks_written'] += 1
self.write_stats['bytes_written'] += len(data)
def get_stats(self):
return self.write_stats
# Setup (reuse NameNode and DataNodes)
nn2 = SimpleNameNode()
datanodes2 = {}
for i in range(1, 4):
dn = DataNodeSimulator(f"dn{i}")
datanodes2[f"dn{i}"] = dn
nn2.register_datanode(f"dn{i}", dn)
# Write file
client = HDFSWriteClient(nn2, datanodes2)
data = "Hello HDFS! " * 10000 # ~120 KB
blocks = client.create_file("/data/hello.txt", data)
print(f"\nWrite completed:")
print(f" Blocks: {client.write_stats['blocks_written']}")
print(f" Bytes: {client.write_stats['bytes_written']}")
# Verify
print(f"\nVerifying blocks:")
for block in blocks:
print(f" {block['id']}: {block['size']} bytes, replicas: {block['locations']}")
# WRITE PATH SUMMARY
print("\n=== Write Path Summary ===")
print(f"{'Step':<25} {'Action':<40} {'Component':<20}")
print("-" * 85)
print(f"{'1. Create':<25} {'Client → NameNode':<40} {'NameNode':<20}")
print(f"{'2. Add block':<25} {'Client → NameNode':<40} {'NameNode':<20}")
print(f"{'3. Pipeline':<25} {'NameNode → Client':<40} {'NameNode':<20}")
print(f"{'4. Write':<25} {'Client → DN1 → DN2 → DN3':<40} {'DataNodes':<20}")
print(f"{'5. ACK':<25} {'DN3 → DN2 → DN1 → Client':<40} {'DataNodes':<20}")
print(f"{'6. Close':<25} {'Client → NameNode':<40} {'NameNode':<20}")
🛡️ Fault Tolerance
🎯 Handling Failures Gracefully
HDFS is designed for fault tolerance: DataNode failures, network partitions, and disk failures are expected. Replication, heartbeats, and automatic recovery ensure data availability.
Fault Tolerance Mechanisms
Replication: 3x by default. Heartbeats: Detect failures. Re-replication: Restore replicas. Checksums: Detect corruption. Pipeline recovery: Handle write failures. HA: NameNode failover.
🛡️ Replication💓 Heartbeats🔄 Re-replication✅ Checksums
# HDFS Fault Tolerance
print("=== Types of Failures ===")
print("""
1. DATANODE FAILURE
- Node crashes
- Disk failure
- Network partition
Impact: Lost replicas
Solution: Re-replicate from other copies
2. NAMENODE FAILURE
- Master crashes
- Single point of failure
Impact: HDFS unavailable
Solution: HA (Active/Standby)
3. NETWORK FAILURE
- Rack failure
- Switch failure
- Partition
Impact: Lost connectivity
Solution: Rack-aware replication
4. DISK FAILURE
- Hard drive failure
- Bad sectors
Impact: Corrupt blocks
Solution: Checksums, re-replication
5. CORRUPTION
- Bit rot
- Software bugs
Impact: Wrong data
Solution: Checksums, block scanner
""")
# HEARTBEAT MECHANISM
print("\n=== Heartbeat Mechanism ===")
print("""
DataNode → NameNode every 3 seconds:
Normal:
DN: heartbeat(node_id, capacity, usage, ...)
NN: ack + commands
Failure detection:
- 10 missed heartbeats (30 sec): stale
- 30 min missed: dead
- NameNode marks blocks for re-replication
Configuration:
dfs.heartbeat.interval = 3 (seconds)
dfs.namenode.heartbeat.recheck-interval = 300000 (ms)
Dead node timeout = 2 * recheck + 10 * heartbeat
= 2 * 300 + 10 * 3
= 600 + 30 = 630 sec (10.5 min)
""")
# RE-REPLICATION
print("\n=== Re-replication ===")
print("""
When a DataNode fails:
1. DETECTION
NameNode marks DataNode as dead
Blocks on that node become under-replicated
2. IDENTIFICATION
NameNode finds blocks with < replication factor
Prioritizes:
- Blocks with 1 replica (critical)
- Blocks with 2 replicas
- Blocks with 3+ replicas
3. REPLICATION
NameNode selects source DataNode
Selects target DataNode (rack-aware)
Sends replication command
4. COMPLETION
Target DataNode copies block
Reports to NameNode
Block map updated
5. PRIORITIZATION
- Critical: 1 replica
- High: 2 replicas
- Normal: 3+ replicas
- Low: over-replicated
""")
# CHECKSUM VERIFICATION
print("\n=== Checksum Verification ===")
print("""
HDFS uses checksums to detect corruption:
1. WRITE TIME
- Client computes checksum (CRC32)
- Sends data + checksum
- DataNode stores both
2. READ TIME
- Client reads data + checksum
- Verifies checksum
- If mismatch: try next replica
3. BLOCK SCANNER
- DataNode scans blocks periodically
- Verifies checksums
- Reports corrupt blocks
4. CORRUPTION HANDLING
- NameNode marks block as corrupt
- Replicates from good replica
- Deletes corrupt block
Configuration:
dfs.datanode.scan.period.hours = 504 (21 days)
dfs.checksum.type = CRC32C
""")
# IMPLEMENTATION
print("\n=== Simulating Fault Tolerance ===")
class FaultTolerantHDFS:
"""Simulate HDFS fault tolerance."""
def __init__(self, replication=3):
self.replication = replication
self.datanodes = {}
self.blocks = {} # block_id -> {locations, replicas}
self.failed_nodes = set()
self.re_rep_events = []
def add_datanode(self, node_id):
self.datanodes[node_id] = {
'id': node_id,
'status': 'alive',
'blocks': set(),
}
def create_block(self, block_id):
"""Create a block with replicas."""
locations = list(self.datanodes.keys())[:self.replication]
self.blocks[block_id] = {
'id': block_id,
'locations': set(locations),
'replicas': self.replication,
}
for loc in locations:
self.datanodes[loc]['blocks'].add(block_id)
def fail_node(self, node_id):
"""Simulate node failure."""
if node_id in self.datanodes:
self.datanodes[node_id]['status'] = 'dead'
self.failed_nodes.add(node_id)
# Find under-replicated blocks
affected_blocks = []
for block_id, block in self.blocks.items():
if node_id in block['locations']:
block['locations'].discard(node_id)
if len(block['locations']) < self.replication:
affected_blocks.append(block_id)
print(f"Node {node_id} failed!")
print(f" Affected blocks: {len(affected_blocks)}")
# Re-replicate
for block_id in affected_blocks:
self._re_replicate(block_id)
def _re_replicate(self, block_id):
"""Re-replicate a block."""
block = self.blocks[block_id]
needed = self.replication - len(block['locations'])
if needed <= 0:
return
# Find available nodes
available = [
n for n in self.datanodes
if self.datanodes[n]['status'] == 'alive'
and n not in block['locations']
]
if len(available) < needed:
print(f" WARNING: Cannot fully re-replicate {block_id}")
needed = len(available)
new_locations = available[:needed]
block['locations'].update(new_locations)
for loc in new_locations:
self.datanodes[loc]['blocks'].add(block_id)
self.re_rep_events.append({
'block': block_id,
'new_locations': new_locations,
'replicas': len(block['locations']),
})
print(f" Re-replicated {block_id} to {new_locations} "
f"(now {len(block['locations'])} replicas)")
def get_status(self):
"""Get cluster status."""
alive = sum(1 for dn in self.datanodes.values() if dn['status'] == 'alive')
dead = sum(1 for dn in self.datanodes.values() if dn['status'] == 'dead')
under_replicated = sum(
1 for b in self.blocks.values()
if len(b['locations']) < self.replication
)
return {
'alive_nodes': alive,
'dead_nodes': dead,
'total_blocks': len(self.blocks),
'under_replicated': under_replicated,
're_rep_events': len(self.re_rep_events),
}
# Simulate
hdfs = FaultTolerantHDFS(replication=3)
# Add 5 DataNodes
for i in range(1, 6):
hdfs.add_datanode(f"dn{i}")
# Create blocks
for i in range(5):
hdfs.create_block(f"blk_{i}")
print("Initial state:")
status = hdfs.get_status()
for key, val in status.items():
print(f" {key}: {val}")
# Fail a node
print()
hdfs.fail_node("dn2")
print("\nAfter failure:")
status = hdfs.get_status()
for key, val in status.items():
print(f" {key}: {val}")
# Show block status
print("\nBlock status:")
for block_id, block in hdfs.blocks.items():
print(f" {block_id}: {len(block['locations'])} replicas {block['locations']}")
# FAULT TOLERANCE SUMMARY
print("\n=== Fault Tolerance Summary ===")
print(f"{'Failure':<25} {'Detection':<25} {'Recovery':<30}")
print("-" * 80)
print(f"{'DataNode':<25} {'Heartbeat (30s)':<25} {'Re-replicate':<30}")
print(f"{'NameNode':<25} {'HA heartbeat':<25} {'Failover':<30}")
print(f"{'Network':<25} {'Timeouts':<25} {'Rack-aware replicas':<30}")
print(f"{'Disk':<25} {'Checksums':<25} {'Re-replicate':<30}")
print(f"{'Corruption':<25} {'Checksums':<25} {'Block scanner':<30}")
🔄 High Availability
🎯 NameNode HA
High Availability (HA) eliminates the NameNode single point of failure. Two NameNodes (Active + Standby) with automatic failover ensure HDFS remains available even if one NameNode fails.
HA Architecture
Active: Serves requests. Standby: Ready to take over. QJM: Shared edit log. ZKFC: Failure detection. Failover: Automatic or manual. Fencing: Prevent split-brain.
🔄 Active/Standby📋 QJM🔍 ZKFC⚡ Auto Failover
# HDFS High Availability
print("=== The Problem ===")
print("""
Without HA:
- Single NameNode
- If it fails → HDFS unavailable
- Downtime until restart
- Manual recovery
Impact:
- Hours of downtime
- Lost productivity
- SLA violations
""")
print("\n=== HA Architecture ===")
print("""
HDFS HA (Hadoop 2.x+):
┌─────────────────┐ ┌─────────────────┐
│ Active │ │ Standby │
│ NameNode │◄───────►│ NameNode │
│ │ │ │
│ - Serves │ │ - Ready │
│ requests │ │ - Syncs edits │
└────────┬────────┘ └────────┬────────┘
│ │
│ ┌─────────────────┐ │
└───►│ JournalNodes │◄───┘
│ (QJM) │
│ Shared edits │
└─────────────────┘
│
│
┌────────┴────────┐
│ ZooKeeper │
│ (Failover) │
└─────────────────┘
Components:
- Active NameNode: Handles all requests
- Standby NameNode: Syncs with Active
- JournalNodes: Shared edit log (QJM)
- ZooKeeper: Failure detection + failover
- ZKFC: ZooKeeper Failover Controller
""")
# QJM
print("\n=== QJM (Quorum Journal Manager) ===")
print("""
QJM provides shared edit log:
Architecture:
- 3+ JournalNodes
- Active NameNode writes edits
- Standby NameNode reads edits
- Quorum: majority must acknowledge
Write path:
Active NN → JournalNode 1
→ JournalNode 2
→ JournalNode 3
(majority ACK)
Read path:
Standby NN ← JournalNode 1
← JournalNode 2
← JournalNode 3
Benefits:
- No shared storage needed
- Fault-tolerant
- Automatic recovery
Configuration:
dfs.namenode.shared.edits.dir = qjournal://jn1:8485;jn2:8485;jn3:8485/mycluster
""")
# FAILOVER
print("\n=== Failover Process ===")
print("""
Automatic failover:
1. ACTIVE FAILS
- Active NameNode crashes
- ZKFC detects failure
2. ZKFC NOTIFIES ZOOKEEPER
- ZKFC on Standby:
- Creates ephemeral znode
- Becomes leader
3. FENCING
- Fence old Active:
- SSH kill
- STONITH
- Disable port
- Prevent split-brain
4. STANDBY → ACTIVE
- Standby reads all edits from QJM
- Applies to namespace
- Becomes Active
5. CLIENTS REDIRECT
- Clients use nameservice ID
- Connect to new Active
- Continue operations
Downtime: Seconds (automatic)
""")
# FENCING
print("\n=== Fencing ===")
print("""
Prevents split-brain (two Active NameNodes):
Methods:
1. SSH FENCING
- SSH to old Active
- Kill NameNode process
2. STONITH
- Shoot The Other Node In The Head
- Power off via IPMI/PDU
3. SHELL FENCING
- Custom script
- Disable network
4. SHARED STORAGE FENCING
- Revoke access
- Prevent writes
Configuration:
dfs.ha.fencing.methods = sshfence
dfs.ha.fencing.ssh.private-key-files = /home/hdfs/.ssh/id_rsa
""")
# IMPLEMENTATION
print("\n=== Simulating HA ===")
import time
import random
class NameNodeHA:
"""Simulate NameNode HA."""
def __init__(self, name):
self.name = name
self.state = 'standby' # 'active', 'standby', 'dead'
self.edits = []
self.last_heartbeat = time.time()
def become_active(self):
self.state = 'active'
print(f" {self.name}: became ACTIVE")
def become_standby(self):
self.state = 'standby'
print(f" {self.name}: became STANDBY")
def fail(self):
self.state = 'dead'
print(f" {self.name}: FAILED")
def process_write(self, op):
if self.state != 'active':
return False
self.edits.append(op)
return True
def sync_edits(self, edits):
if self.state == 'standby':
self.edits = edits.copy()
return True
return False
class JournalNode:
"""Simulate QJM JournalNode."""
def __init__(self, name):
self.name = name
self.edits = []
self.alive = True
def write(self, edit):
if self.alive:
self.edits.append(edit)
return True
return False
def read(self):
return self.edits.copy()
class ZKFC:
"""Simulate ZooKeeper Failover Controller."""
def __init__(self):
self.leader = None
def check_failover(self, active, standby, journal):
"""Check if failover needed."""
if active.state == 'dead':
print("\nZKFC: Active failed! Initiating failover...")
# Fence old active
print(" Fencing old active...")
# Promote standby
standby.become_active()
self.leader = standby
return True
# Sync edits
if standby.state == 'standby':
standby.sync_edits(journal.read())
return False
# Simulate
print("=== HA Simulation ===")
# Create components
nn1 = NameNodeHA("nn1")
nn2 = NameNodeHA("nn2")
jn1 = JournalNode("jn1")
jn2 = JournalNode("jn2")
jn3 = JournalNode("jn3")
zkfc = ZKFC()
# Initial state
nn1.become_active()
nn2.become_standby()
print(f"\nInitial: {nn1.name}={nn1.state}, {nn2.name}={nn2.state}")
# Process writes
print("\nProcessing writes:")
for i in range(5):
op = f"write_{i}"
if nn1.process_write(op):
# Write to journal
for jn in [jn1, jn2, jn3]:
jn.write(op)
print(f" {op}: written by {nn1.name}")
# Check failover (no failure)
zkfc.check_failover(nn1, nn2, jn1)
# Fail active
print("\n--- Simulating failure ---")
nn1.fail()
# Check failover
zkfc.check_failover(nn1, nn2, jn1)
print(f"\nAfter failover: {nn1.name}={nn1.state}, {nn2.name}={nn2.state}")
# Continue writes
print("\nContinuing writes:")
for i in range(3):
op = f"write_after_{i}"
if nn2.process_write(op):
for jn in [jn1, jn2, jn3]:
jn.write(op)
print(f" {op}: written by {nn2.name}")
# HA SUMMARY
print("\n=== HA Summary ===")
print(f"{'Component':<25} {'Role':<40}")
print("-" * 65)
print(f"{'Active NameNode':<25} {'Serves requests':<40}")
print(f"{'Standby NameNode':<25} {'Ready, syncs edits':<40}")
print(f"{'JournalNodes (QJM)':<25} {'Shared edit log':<40}")
print(f"{'ZooKeeper':<25} {'Failure detection':<40}")
print(f"{'ZKFC':<25} {'Failover controller':<40}")
print(f"{'Fencing':<25} {'Prevent split-brain':<40}")
# FEDERATION
print("\n=== Federation ===")
print("""
Federation (Hadoop 2.x+):
Multiple NameNodes, each managing a namespace:
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ NameNode 1 │ │ NameNode 2 │ │ NameNode 3 │
│ Namespace 1 │ │ Namespace 2 │ │ Namespace 3 │
└──────┬──────┘ └──────┬──────┘ └──────┬──────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────────────────────────┐
│ DataNodes │
│ (shared storage) │
└─────────────────────────────────────────────┘
Benefits:
- Scale namespace
- Isolate workloads
- Higher availability
Configuration:
dfs.nameservices = ns1,ns2
dfs.ha.namenodes.ns1 = nn1,nn2
dfs.namenode.rpc-address.ns1.nn1 = host1:8020
""")
⌨️ Shell Commands
🎯 HDFS Command Line
The HDFS shell provides commands to interact with HDFS. These commands are similar to Unix file commands but operate on the distributed file system.
Essential Commands
ls: List files. put: Upload. get: Download. mkdir: Create directory. rm: Delete. cat: View. du: Disk usage. fsck: Check health.
⌨️ hdfs dfs📁 File ops📊 Admin🔍 Health
# HDFS Shell Commands
print("=== Basic File Operations ===")
print("""
# List files
hdfs dfs -ls / # List root
hdfs dfs -ls -R /data # Recursive
hdfs dfs -ls -h /data # Human-readable
# Create directory
hdfs dfs -mkdir /data # Single
hdfs dfs -mkdir -p /data/logs # Create parents
# Upload files
hdfs dfs -put local.txt /data/ # Upload
hdfs dfs -copyFromLocal local.txt /data/ # Same
hdfs dfs -moveFromLocal local.txt /data/ # Move
# Download files
hdfs dfs -get /data/file.txt . # Download
hdfs dfs -copyToLocal /data/file.txt . # Same
hdfs dfs -getmerge /data/logs merged.txt # Merge
# View files
hdfs dfs -cat /data/file.txt # Print
hdfs dfs -tail /data/file.txt # Last 1KB
hdfs dfs -head /data/file.txt # First 1KB
# Delete
hdfs dfs -rm /data/file.txt # Delete file
hdfs dfs -rm -r /data/logs # Recursive
hdfs dfs -rm -skipTrash /data/file.txt # Permanent
""")
print("\n=== Advanced Operations ===")
print("""
# Copy within HDFS
hdfs dfs -cp /data/a.txt /backup/ # Copy
hdfs dfs -mv /data/a.txt /backup/ # Move
# Permissions
hdfs dfs -chmod 755 /data/file.txt
hdfs dfs -chown user:group /data/file.txt
# Disk usage
hdfs dfs -du /data # Size of files
hdfs dfs -du -h /data # Human-readable
hdfs dfs -dus /data # Summary
# Count
hdfs dfs -count /data # Count files/dirs
hdfs dfs -count -q /data # With quotas
# Set replication
hdfs dfs -setrep -w 5 /data/file.txt
# Trash
hdfs dfs -expunge # Empty trash
hdfs dfs -ls /user/hdfs/.Trash # View trash
""")
print("\n=== Admin Commands ===")
print("""
# Cluster report
hdfs dfsadmin -report # Cluster status
hdfs dfsadmin -report -live # Live nodes
hdfs dfsadmin -report -dead # Dead nodes
# Safe mode
hdfs dfsadmin -safemode get # Check
hdfs dfsadmin -safemode enter # Enter
hdfs dfsadmin -safemode leave # Leave
hdfs dfsadmin -safemode wait # Wait
# Refresh
hdfs dfsadmin -refreshNodes # Refresh nodes
hdfs dfsadmin -refreshUserToGroupsMappings
# Save namespace
hdfs dfsadmin -saveNamespace # Save fsimage
# Rolling upgrade
hdfs dfsadmin -rollingUpgrade prepare
hdfs dfsadmin -rollingUpgrade query
hdfs dfsadmin -rollingUpgrade finalize
# HA
hdfs haadmin -getServiceState nn1
hdfs haadmin -failover nn1 nn2
hdfs haadmin -transitionToActive nn1
hdfs haadmin -transitionToStandby nn1
# Federation
hdfs dfsadmin -router
hdfs dfsadmin -getRouterRpcAddress
""")
print("\n=== Health Check Commands ===")
print("""
# File system check
hdfs fsck / # Check root
hdfs fsck /data/file.txt # Check file
hdfs fsck /data -files -blocks # Detailed
hdfs fsck /data -locations # Block locations
hdfs fsck /data -move # Move corrupt to lost+found
hdfs fsck /data -delete # Delete corrupt
# Output example:
# /data/file.txt 500 MB, 4 blocks
# Block 0: 128 MB, 3 replicas
# ...
# Status: HEALTHY
# Total size: 500 MB
# Total blocks: 4
# Total replicas: 12
# Under-replicated: 0
# Corrupt: 0
# Missing: 0
# Block scanner
hdfs fsck /data -blocks -locations
hdfs fsck /data -openforwrite # Files being written
""")
print("\n=== Special Commands ===")
print("""
# Find files
hdfs dfs -find /data -name "*.log"
hdfs dfs -find /data -size +100M
# Text processing
hdfs dfs -text /data/file.gz # Decompress
hdfs dfs -text /data/file.orc # ORC format
# Checksum
hdfs dfs -checksum /data/file.txt
# Get file status
hdfs dfs -stat /data/file.txt
hdfs dfs -stat "%b %n %o" /data/file.txt
# Truncate
hdfs dfs -truncate -w 100 /data/file.txt
# Concatenate
hdfs dfs -cat /data/a.txt /data/b.txt > combined.txt
""")
# COMMAND EXAMPLES
print("\n=== Practical Examples ===")
print("""
# Upload a directory
hdfs dfs -put /local/data /hdfs/data
# Download multiple files
hdfs dfs -get /data/*.txt ./
# Copy between clusters
hadoop distcp hdfs://cluster1/data hdfs://cluster2/data
# Set replication for a directory
hdfs dfs -setrep -w 2 -R /data
# Check disk usage by directory
hdfs dfs -du -h -s /data/*
""")
# PERMISSIONS
print("\n=== Permissions ===")
print("""
HDFS permissions are similar to Unix:
rwx rwx rwx
owner group others
Commands:
hdfs dfs -chmod 755 /data/file.txt
hdfs dfs -chmod u+x /data/file.txt
hdfs dfs -chmod g-w /data/file.txt
hdfs dfs -chmod o-rwx /data/file.txt
hdfs dfs -chown user /data/file.txt
hdfs dfs -chown user:group /data/file.txt
hdfs dfs -chgrp group /data/file.txt
Sticky bit:
hdfs dfs -chmod +t /data/tmp
ACLs:
hdfs dfs -setfacl -m user:alice:rwx /data/file.txt
hdfs dfs -getfacl /data/file.txt
""")
# SUMMARY TABLE
print("\n=== Command Summary ===")
print(f"{'Category':<20} {'Commands':<50}")
print("-" * 70)
print(f"{'File ops':<20} {'ls, mkdir, put, get, rm, cp, mv':<50}")
print(f"{'Viewing':<20} {'cat, tail, head, text':<50}")
print(f"{'Disk usage':<20} {'du, dus, count':<50}")
print(f"{'Admin':<20} {'dfsadmin, haadmin':<50}")
print(f"{'Health':<20} {'fsck, report':<50}")
print(f"{'Permissions':<20} {'chmod, chown, chgrp, setfacl':<50}")
print(f"{'Replication':<20} {'setrep':<50}")
💻 Java/Python API
🎯 Programming HDFS
HDFS provides Java (native) and Python (via libraries) APIs for programmatic access. These APIs are used by MapReduce, Spark, Hive, and custom applications.
API Options
Java: Native API. Python: hdfs, pyarrow. REST: WebHDFS. CLI: Shell commands. FUSE: Mount as filesystem. Libhdfs: C API.
☕ Java🐍 Python🌐 REST🔌 FUSE
# HDFS APIs
print("=== Java API ===")
print("""
Native HDFS API (org.apache.hadoop.fs):
// Configuration
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://namenode:8020");
// Get FileSystem
FileSystem fs = FileSystem.get(conf);
// Create file
Path path = new Path("/data/file.txt");
FSDataOutputStream out = fs.create(path);
out.writeBytes("Hello HDFS!");
out.close();
// Read file
FSDataInputStream in = fs.open(path);
BufferedReader reader = new BufferedReader(new InputStreamReader(in));
String line;
while ((line = reader.readLine()) != null) {
System.out.println(line);
}
reader.close();
// List files
FileStatus[] statuses = fs.listStatus(new Path("/data"));
for (FileStatus status : statuses) {
System.out.println(status.getPath());
}
// Delete
fs.delete(path, false);
// Close
fs.close();
""")
print("\n=== Python API (hdfs library) ===")
print("""
Install: pip install hdfs
from hdfs import InsecureClient
# Connect
client = InsecureClient('http://namenode:9870', user='hdfs')
# List files
files = client.list('/data')
print(files)
# Write file
with client.write('/data/file.txt', encoding='utf-8') as writer:
writer.write('Hello HDFS!')
# Read file
with client.read('/data/file.txt', encoding='utf-8') as reader:
content = reader.read()
print(content)
# Upload local file
client.upload('/data', 'local_file.txt')
# Download
client.download('/data/file.txt', 'local_copy.txt')
# Delete
client.delete('/data/file.txt')
# Get file status
status = client.status('/data/file.txt')
print(status)
""")
print("\n=== Python API (PyArrow) ===")
print("""
Install: pip install pyarrow
import pyarrow.fs as fs
# Connect
hdfs = fs.HadoopFileSystem('namenode', port=8020)
# Read file
with hdfs.open_input_file('/data/file.txt') as f:
data = f.read()
print(data)
# Write file
with hdfs.open_output_file('/data/output.txt') as f:
f.write(b'Hello HDFS!')
# List directory
selector = fs.FileSelector('/data', recursive=True)
files = hdfs.get_file_info(selector)
for file in files:
print(file.path, file.size)
# Delete
hdfs.delete_file('/data/file.txt')
""")
print("\n=== WebHDFS REST API ===")
print("""
HTTP REST API for HDFS:
Base URL: http://namenode:9870/webhdfs/v1
# List files
GET /webhdfs/v1/data?op=LISTSTATUS
# Get file
GET /webhdfs/v1/data/file.txt?op=OPEN
# Create file
PUT /webhdfs/v1/data/file.txt?op=CREATE
# Append
POST /webhdfs/v1/data/file.txt?op=APPEND
# Delete
DELETE /webhdfs/v1/data/file.txt?op=DELETE
# Rename
PUT /webhdfs/v1/data/file.txt?op=RENAME&destination=/data/new.txt
# Get file status
GET /webhdfs/v1/data/file.txt?op=GETFILESTATUS
# Mkdir
PUT /webhdfs/v1/data/newdir?op=MKDIRS
Example with curl:
curl -i "http://namenode:9870/webhdfs/v1/data?op=LISTSTATUS"
curl -i -X PUT "http://namenode:9870/webhdfs/v1/data/file.txt?op=CREATE"
""")
print("\n=== FUSE Mount ===")
print("""
Mount HDFS as a local filesystem:
# Install
sudo apt-get install hadoop-hdfs-fuse
# Mount
mkdir /mnt/hdfs
hadoop-fuse-dfs hdfs://namenode:8020 /mnt/hdfs
# Use like local FS
ls /mnt/hdfs/data
cat /mnt/hdfs/data/file.txt
cp /mnt/hdfs/data/file.txt /tmp/
# Unmount
fusermount -u /mnt/hdfs
Benefits:
- Familiar file interface
- Works with any tool
- No code changes
Limitations:
- Higher latency
- Not for heavy I/O
""")
# IMPLEMENTATION EXAMPLE
print("\n=== Complete Python Example ===")
print("""
from hdfs import InsecureClient
import json
class HDFSManager:
def __init__(self, url, user='hdfs'):
self.client = InsecureClient(url, user=user)
def write_json(self, path, data):
\"\"\"Write JSON to HDFS.\"\"\"
with self.client.write(path, encoding='utf-8') as writer:
json.dump(data, writer)
def read_json(self, path):
\"\"\"Read JSON from HDFS.\"\"\"
with self.client.read(path, encoding='utf-8') as reader:
return json.load(reader)
def list_files(self, path='/'):
\"\"\"List files in directory.\"\"\"
return self.client.list(path)
def upload_dir(self, local_dir, hdfs_dir):
\"\"\"Upload entire directory.\"\"\"
self.client.upload(hdfs_dir, local_dir, overwrite=True)
def download_dir(self, hdfs_dir, local_dir):
\"\"\"Download entire directory.\"\"\"
self.client.download(hdfs_dir, local_dir, overwrite=True)
def delete(self, path, recursive=False):
\"\"\"Delete file or directory.\"\"\"
self.client.delete(path, recursive=recursive)
def get_size(self, path):
\"\"\"Get file size.\"\"\"
status = self.client.status(path)
return status['length']
# Usage
if __name__ == '__main__':
hdfs = HDFSManager('http://namenode:9870')
# Write
data = {'name': 'Alice', 'age': 30}
hdfs.write_json('/data/user.json', data)
# Read
loaded = hdfs.read_json('/data/user.json')
print(loaded)
# List
files = hdfs.list_files('/data')
print(files)
# Size
size = hdfs.get_size('/data/user.json')
print(f'Size: {size} bytes')
""")
# BEST PRACTICES
print("\n=== Best Practices ===")
print("""
1. REUSE FILESYSTEM OBJECT
- Don't create for each operation
- Expensive to create
- Cache and reuse
2. CLOSE RESOURCES
- Use try-finally
- Or context managers
- Prevent leaks
3. BATCH OPERATIONS
- Use listStatus for multiple files
- Reduce RPC calls
4. BUFFERING
- Use BufferedInputStream/OutputStream
- Reduce I/O operations
5. ERROR HANDLING
- Handle IOException
- Retry on failure
- Log errors
6. CONFIGURATION
- Use Configuration object
- Set timeouts
- Set replication
7. COMPRESSION
- Use compression codecs
- Reduce storage/network
- Snappy, Gzip
8. SECURITY
- Use Kerberos
- Set permissions
- Use tokens
""")
# API SUMMARY
print("\n=== API Summary ===")
print(f"{'API':<20} {'Language':<15} {'Best For':<30}")
print("-" * 65)
print(f"{'Java API':<20} {'Java':<15} {'Native, full control':<30}")
print(f"{'hdfs (Python)':<20} {'Python':<15} {'Simplicity':<30}")
print(f"{'PyArrow':<20} {'Python':<15} {'Data science':<30}")
print(f"{'WebHDFS':<20} {'REST':<15} {'Any language':<30}")
print(f"{'FUSE':<20} {'Any':<15} {'Filesystem access':<30}")
print(f"{'libhdfs':<20} {'C':<15} {'C/C++ apps':<30}")
🌐 Hadoop Ecosystem
🎯 HDFS in Context
HDFS is the storage layer of the Hadoop ecosystem. It integrates with many tools: MapReduce, Spark, Hive, HBase, Kafka, and more. Understanding this ecosystem is key to Big Data architecture.
Ecosystem Components
Storage: HDFS. Processing: MapReduce, Spark. SQL: Hive, Impala. NoSQL: HBase. Streaming: Kafka, Flink. Ingestion: Sqoop, Flume.
🌐 Ecosystem💾 HDFS⚡ Spark🐘 Hive
# Hadoop Ecosystem
print("=== Hadoop Ecosystem Diagram ===")
print("""
┌─────────────────────────────────────────────────────────────┐
│ HADOOP ECOSYSTEM │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Hive │ │ Pig │ │ Spark │ │ Flink │ │
│ │ (SQL) │ │ (Script)│ │ (Memory)│ │ (Stream)│ │
│ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │ │
│ ┌────┴────────────┴────────────┴────────────┴────┐ │
│ │ MapReduce / YARN │ │
│ │ (Resource Management + Batch) │ │
│ └─────────────────────┬───────────────────────────┘ │
│ │ │
│ ┌─────────────────────┴───────────────────────────┐ │
│ │ HDFS │ │
│ │ (Distributed File System) │ │
│ └─────────────────────┬───────────────────────────┘ │
│ │ │
│ ┌─────────────────────┴───────────────────────────┐ │
│ │ Storage Layer │ │
│ │ (Local Disks on DataNodes) │ │
│ └─────────────────────────────────────────────────┘ │
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Sqoop │ │ Flume │ │ Kafka │ │ HBase │ │
│ │(RDBMS) │ │ (Logs) │ │(Stream) │ │ (NoSQL) │ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
""")
# COMPONENTS
print("\n=== Ecosystem Components ===")
print("""
1. HDFS (Storage)
- Distributed file system
- Fault-tolerant
- Scalable
2. YARN (Resource Management)
- Cluster resource manager
- Schedules jobs
- Multi-tenant
3. MapReduce (Batch Processing)
- Original Hadoop processing
- Disk-based
- Fault-tolerant
4. Spark (In-Memory Processing)
- 100x faster than MapReduce
- Batch + streaming + ML
- Most popular
5. Hive (SQL)
- SQL on Hadoop
- Data warehouse
- Batch queries
6. Pig (Scripting)
- Data flow language
- ETL
- Less popular now
7. HBase (NoSQL)
- Column-family store
- Random read/write
- On top of HDFS
8. Kafka (Streaming)
- Distributed messaging
- Real-time
- Pub/sub
9. Sqoop (RDBMS Import)
- Import from databases
- Export to databases
- Batch
10. Flume (Log Collection)
- Collect logs
- Aggregate
- Move to HDFS
11. ZooKeeper (Coordination)
- Distributed coordination
- Configuration
- Leader election
12. Oozie (Workflow)
- Job orchestration
- DAG workflows
- Scheduling
""")
# INTEGRATION
print("\n=== Integration Patterns ===")
print("""
1. BATCH PROCESSING
Data → HDFS → MapReduce/Spark → HDFS → Hive → BI
2. REAL-TIME
Data → Kafka → Spark Streaming/Flink → HDFS/HBase
3. SQL ANALYTICS
HDFS → Hive/Impala/Presto → BI tools
4. NoSQL
HBase on HDFS → Random access
5. MACHINE LEARNING
HDFS → Spark MLlib → Models
6. ETL
RDBMS → Sqoop → HDFS → Transform → HDFS
""")
# COMMON STACKS
print("\n=== Common Stacks ===")
print("""
1. CLASSIC HADOOP
HDFS + MapReduce + YARN + Hive
2. MODERN HADOOP
HDFS + Spark + YARN + Hive + Kafka
3. CLOUD
S3 + EMR/Spark + Glue + Athena
(HDFS replaced by S3)
4. STREAMING
Kafka + Flink + HDFS/HBase
5. DATA SCIENCE
HDFS + Spark + Jupyter + MLflow
""")
# IMPLEMENTATION
print("\n=== Simulating Ecosystem ===")
class HadoopEcosystem:
"""Simulate Hadoop ecosystem."""
def __init__(self):
self.hdfs = {} # Simulated HDFS
self.jobs = []
def hdfs_put(self, path, data):
self.hdfs[path] = data
print(f"HDFS: Stored {path}")
def hdfs_get(self, path):
return self.hdfs.get(path)
def mapreduce_job(self, input_path, output_path, mapper, reducer):
"""Simulate MapReduce job."""
print(f"MapReduce: Processing {input_path}")
# Read input
data = self.hdfs_get(input_path)
if not data:
print(" Input not found")
return
# Map
mapped = mapper(data)
print(f" Mapped: {len(mapped)} records")
# Shuffle & Sort
from collections import defaultdict
grouped = defaultdict(list)
for key, value in mapped:
grouped[key].append(value)
# Reduce
reduced = []
for key, values in grouped.items():
reduced.append(reducer(key, values))
# Write output
self.hdfs_put(output_path, reduced)
print(f" Reduced: {len(reduced)} records")
self.jobs.append({
'type': 'mapreduce',
'input': input_path,
'output': output_path,
})
return reduced
def spark_job(self, input_path, transform):
"""Simulate Spark job."""
print(f"Spark: Processing {input_path}")
data = self.hdfs_get(input_path)
result = transform(data)
print(f" Result: {len(result)} records")
return result
def hive_query(self, table, query):
"""Simulate Hive query."""
print(f"Hive: Querying {table}")
data = self.hdfs.get(table, [])
result = query(data)
print(f" Result: {len(result)} rows")
return result
# Example
eco = HadoopEcosystem()
# Store data in HDFS
data = [
('apple', 1), ('banana', 2), ('apple', 3),
('banana', 4), ('cherry', 5), ('apple', 6),
]
eco.hdfs_put('/data/fruits', data)
# MapReduce: word count
def mapper(data):
return data # Already key-value
def reducer(key, values):
return (key, sum(values))
result = eco.mapreduce_job('/data/fruits', '/data/counts', mapper, reducer)
print(f" Output: {result}")
# Spark: filter
spark_result = eco.spark_job('/data/fruits', lambda d: [x for x in d if x[1] > 2])
print(f" Filtered: {spark_result}")
# Hive: aggregate
hive_result = eco.hive_query('/data/fruits', lambda d: sorted(d, key=lambda x: x[1]))
print(f" Sorted: {hive_result}")
# ECOSYSTEM SUMMARY
print("\n=== Ecosystem Summary ===")
print(f"{'Layer':<20} {'Components':<40}")
print("-" * 60)
print(f"{'Storage':<20} {'HDFS, HBase':<40}")
print(f"{'Resource Mgmt':<20} {'YARN':<40}")
print(f"{'Batch':<20} {'MapReduce, Spark, Pig':<40}")
print(f"{'SQL':<20} {'Hive, Impala, Presto':<40}")
print(f"{'Streaming':<20} {'Kafka, Flink, Storm':<40}")
print(f"{'Ingestion':<20} {'Sqoop, Flume':<40}")
print(f"{'Coordination':<20} {'ZooKeeper':<40}")
print(f"{'Workflow':<20} {'Oozie, Airflow':<40}")
⚖️ HDFS vs Others
🎯 Comparing Storage Systems
HDFS is one of many distributed storage systems. Comparing it with alternatives (S3, Ceph, GlusterFS, Cassandra) helps choose the right tool for the job.
Comparison
HDFS: Hadoop storage. S3: Object store. Ceph: Unified storage. GlusterFS: Scale-out NAS. Cassandra: NoSQL DB. GFS: Google predecessor.
⚖️ HDFS⚖️ S3⚖️ Ceph⚖️ Cassandra
# HDFS vs Other Storage Systems
print("=== Comparison Table ===")
print(f"{'Feature':<25} {'HDFS':<20} {'S3':<20} {'Ceph':<20} {'Cassandra':<20}")
print("-" * 105)
print(f"{'Type':<25} {'Distributed FS':<20} {'Object store':<20} {'Unified':<20} {'NoSQL DB':<20}")
print(f"{'Consistency':<25} {'Strong':<20} {'Eventual':<20} {'Strong':<20} {'Tunable':<20}")
print(f"{'Latency':<25} {'High':<20} {'Medium':<20} {'Low':<20} {'Low':<20}")
print(f"{'Throughput':<25} {'Very high':<20} {'High':<20} {'High':<20} {'High':<20}")
print(f"{'File size':<25} {'Large':<20} {'Any':<20} {'Any':<20} {'Small':<20}")
print(f"{'Random write':<25} {'No':<20} {'Yes':<20} {'Yes':<20} {'Yes':<20}")
print(f"{'POSIX':<25} {'Partial':<20} {'No':<20} {'Yes':<20} {'No':<20}")
print(f"{'Cost':<25} {'Low':<20} {'Pay-per-use':<20} {'Low':<20} {'Medium':<20}")
print(f"{'Scale':<25} {'PB':<20} {'Unlimited':<20} {'PB':<20} {'PB':<20}")
print(f"{'Use case':<25} {'Batch':<20} {'Cloud':<20} {'Unified':<20} {'OLTP':<20}")
""")
print("\n=== HDFS vs S3 ===")
print("""
HDFS:
- Distributed file system
- Strong consistency
- POSIX-like
- Runs on commodity hardware
- High throughput for large files
- Best for: Hadoop/Spark workloads
S3:
- Object storage
- Eventual consistency (now strong)
- HTTP API
- Managed service
- Unlimited scale
- Best for: Cloud-native apps
When to use HDFS:
- On-premise cluster
- Data locality needed
- Hadoop ecosystem
- Cost control
When to use S3:
- Cloud-native
- Serverless
- Unlimited scale
- Pay-per-use
""")
print("\n=== HDFS vs Ceph ===")
print("""
HDFS:
- File system
- Optimized for large files
- Batch processing
- Hadoop ecosystem
Ceph:
- Unified storage (object, block, file)
- RADOS underlying
- Self-healing
- No single point of failure
- OpenStack integration
When to use HDFS:
- Hadoop workloads
- Batch processing
When to use Ceph:
- Unified storage
- OpenStack
- Block storage
- Object storage
""")
print("\n=== HDFS vs GlusterFS ===")
print("""
HDFS:
- Master-slave (NameNode)
- Block-based
- Optimized for large files
- Batch processing
GlusterFS:
- Peer-to-peer (no master)
- File-based
- Scale-out NAS
- POSIX compliant
- General purpose
When to use HDFS:
- Hadoop/Spark
- Big Data batch
When to use GlusterFS:
- General file storage
- POSIX compliance
- No single point
""")
print("\n=== HDFS vs Cassandra ===")
print("""
HDFS:
- File system
- Large files
- Batch processing
- Sequential access
Cassandra:
- NoSQL database
- Wide-column store
- Random read/write
- Real-time
- No single point
When to use HDFS:
- Batch analytics
- Data lake
When to use Cassandra:
- Real-time applications
- Time-series
- High availability
- Low latency
""")
print("\n=== HDFS vs GFS ===")
print("""
GFS (Google File System):
- Predecessor to HDFS
- Similar architecture
- Single master
- Chunk-based (64 MB)
- Proprietary
HDFS:
- Open source
- Inspired by GFS
- Block-based (128 MB)
- Hadoop ecosystem
Relationship:
- HDFS is open-source GFS
- GFS paper (2003) → Hadoop
""")
print("\n=== HDFS vs Object Stores ===")
print("""
Object Stores (S3, Azure Blob, GCS):
- Objects, not files
- HTTP API
- Eventually consistent
- Unlimited scale
- Pay-per-use
HDFS:
- Files, not objects
- POSIX-like API
- Strong consistency
- Limited scale
- Fixed cost
Modern trend:
- Cloud: S3 + EMR/Spark
- On-prem: HDFS
- Hybrid: HDFS + S3 gateway
""")
# DECISION MATRIX
print("\n=== Decision Matrix ===")
print(f"{'Requirement':<25} {'Best Choice':<25} {'Alternative':<25}")
print("-" * 75)
print(f"{'Hadoop batch':<25} {'HDFS':<25} {'S3 (cloud)':<25}")
print(f"{'Real-time OLTP':<25} {'Cassandra':<25} {'HBase':<25}")
print(f"{'Object storage':<25} {'S3':<25} {'Ceph':<25}")
print(f"{'POSIX file':<25} {'GlusterFS':<25} {'Ceph':<25}")
print(f"{'Unified storage':<25} {'Ceph':<25} {'HDFS + HBase':<25}")
print(f"{'Cloud-native':<25} {'S3':<25} {'GCS, Azure':<25}")
print(f"{'On-premise':<25} {'HDFS':<25} {'Ceph':<25}")
print(f"{'Low latency':<25} {'Cassandra':<25} {'HBase':<25}")
print(f"{'High throughput':<25} {'HDFS':<25} {'S3':<25}")
# SUMMARY
print("\n=== Summary ===")
print("""
HDFS strengths:
✓ Open source
✓ Hadoop ecosystem
✓ High throughput
✓ Cost-effective
✓ Fault-tolerant
HDFS weaknesses:
✗ Single NameNode (mitigated by HA)
✗ High latency
✗ No random writes
✗ Not POSIX compliant
✗ On-premise focus
Choose HDFS when:
- Hadoop/Spark workloads
- Batch processing
- Large files
- On-premise cluster
- Cost control
Choose alternatives when:
- Cloud-native (S3)
- Real-time (Cassandra)
- Unified storage (Ceph)
- POSIX (GlusterFS)
""")
🚀 Future: HDFS in Modern Computing
🎯 Where HDFS Is Going
HDFS is evolving: cloud integration, erasure coding, federation, and new architectures. While cloud object stores challenge HDFS, it remains essential for on-premise Big Data.
Future Directions
Cloud: S3 integration. Erasure coding: Lower overhead. Federation: Scale namespace. NVMe: Faster storage. Kubernetes: Container-native. Lakehouse: Unified analytics.
☁️ Cloud🧬 Erasure Coding🌐 Federation⚡ NVMe
# Future of HDFS
print("=== 1. Cloud Integration ===")
print("""
Trend: HDFS + Cloud
Approaches:
1. HDFS on Cloud
- Run HDFS on cloud VMs
- EMR, Dataproc, HDInsight
- Managed Hadoop
2. S3 as HDFS
- S3A connector
- Replace HDFS with S3
- Cloud-native
3. Hybrid
- HDFS on-prem
- S3 for overflow
- Data tiering
Future:
- Cloud-native Hadoop
- S3-first architectures
- HDFS for on-prem only
""")
print("\n=== 2. Erasure Coding ===")
print("""
Trend: Replace replication with erasure coding
Current: 3x replication
- 3 copies of every block
- 200% overhead
- Simple, fast
Erasure Coding (HDFS-EC):
- Reed-Solomon codes
- k data blocks + m parity blocks
- Example: 6+3 (50% overhead)
- Much lower storage cost
Benefits:
- 50% storage savings
- Same fault tolerance
- Better for cold data
Challenges:
- More CPU for encoding
- Reconstruction cost
- Not for hot data
Configuration:
dfs.replication = 1 (for EC)
hdfs ec -enablePolicy -policy RS-6-3-1024k
hdfs ec -setPolicy -path /data/cold -policy RS-6-3-1024k
""")
print("\n=== 3. Federation ===")
print("""
Trend: Multiple NameNodes
Problem:
- Single NameNode memory limit
- ~1 GB per million blocks
- 100M files = ~30 GB RAM
Federation:
- Multiple NameNodes
- Each manages namespace
- Shared DataNodes
- Block pool IDs
Benefits:
- Scale namespace
- Isolate workloads
- Higher availability
Configuration:
dfs.nameservices = ns1,ns2
dfs.ha.namenodes.ns1 = nn1,nn2
dfs.namenode.rpc-address.ns1.nn1 = host1:8020
Future:
- Router-based federation
- Transparent access
- Global namespace
""")
print("\n=== 4. NVMe and Storage ===")
print("""
Trend: Faster storage
Current:
- HDD (spinning disks)
- High throughput
- High latency
NVMe:
- Solid-state drives
- Much lower latency
- Higher IOPS
- Higher cost
HDFS with NVMe:
- Better random I/O
- Lower latency
- New use cases
Storage tiers:
- Hot: NVMe SSD
- Warm: SATA SSD
- Cold: HDD
- Archive: Tape/cloud
HDFS supports:
- Storage policies
- Tiered storage
- Heterogeneous storage
""")
print("\n=== 5. Kubernetes and Containers ===")
print("""
Trend: Container-native HDFS
Approaches:
1. HDFS on Kubernetes
- StatefulSets
- Persistent volumes
- Helm charts
2. Container-native storage
- Rook (Ceph)
- Portworx
- OpenEBS
3. Serverless
- No cluster management
- Pay per use
Benefits:
- Elastic scaling
- Resource isolation
- DevOps friendly
Challenges:
- Data locality
- Persistent storage
- Network performance
Future:
- Kubernetes-native HDFS
- CSI drivers
- Operator patterns
""")
print("\n=== 6. Lakehouse Architecture ===")
print("""
Trend: Unify data lake + warehouse
Components:
- Storage: HDFS or S3
- Table format: Iceberg, Delta, Hudi
- Compute: Spark, Presto, Flink
- Catalog: Hive Metastore, Nessie
Benefits:
- ACID transactions
- Schema evolution
- Time travel
- Unified batch + streaming
HDFS role:
- Storage layer
- Or replaced by S3
Future:
- Open table formats
- Multi-engine access
- Unified governance
""")
print("\n=== 7. New Frontiers ===")
print("""
1. QUANTUM STORAGE
- Quantum computing
- New storage models
- Experimental
2. 5G/6G
- Massive IoT
- Edge storage
- Low latency
3. AI/ML
- Feature stores
- Model storage
- Training data
4. EDGE
- Edge HDFS
- Federated
- Local processing
5. SERVERLESS
- No infrastructure
- Pay per use
- Auto-scaling
6. GREEN COMPUTING
- Energy efficiency
- Carbon-aware
- Sustainable
""")
# TIMELINE
print("\n=== Timeline ===")
print("""
Past:
2003: GFS paper (Google)
2006: Hadoop (Yahoo)
2008: Hadoop summit
2010: HDFS 0.20
2012: Hadoop 2.0 (YARN, HA)
2017: Hadoop 3.0 (Erasure coding)
Present:
2024: Cloud integration
2025: Lakehouse, Kubernetes
Future:
2030: Quantum storage?
2035: Edge HDFS?
2040: New paradigms?
Beyond: Unknown
""")
print("\n=== Final Thoughts ===")
print("""
HDFS is:
- Foundational for Big Data
- Proven at scale
- Evolving with cloud
- Still relevant
The future is:
- Cloud-native
- Erasure coded
- Federated
- Containerized
- Unified (lakehouse)
"Data is the new oil,
and HDFS is the pipeline."
Keep exploring. Keep building.
""")