Distributed File Systems (HDFS)

Why sorting 10TB wants colocated blocks instead of remote blobs

The picker ended asking what splits files into 128MB replicated blocks so compute moves to the data. You must sort 10TB of logs with MapReduce, meaning a framework that ships code to data splits, and striped block volumes cannot help because one filesystem still feeds one machine. Object storage holds immutable blobs well, but its listing and rename are non-atomic and throughput per prefix throttles. HDFS, meaning the Hadoop Distributed File System that splits files into large replicated blocks with compute beside them, was built for write-once parallel sequential scans: large files, rack-aware replication, and mappers reading local disks.

Client ──► NameNode (metadata: file → [block1@DN1,DN2,DN3])
                │
                ▼
   block = 128MB, replicated 3× across racks
   DataNodes heartbeat, report blocks
   Read: ask NameNode → connect directly to nearest DataNode

The NameNode, meaning the master holding the namespace and block locations in RAM, is the single metadata hop. DataNodes, meaning workers storing blocks and heartbeating health, serve bytes directly once the client learns locations. That split is the whole design: tiny metadata centralized, huge data decentralized.

How namespace, pipelines, and rack placement divide the work

Naive replication scatters copies randomly, which survives disk loss but loses a whole rack to one switch fire. HDFS places replica one on the local rack, replica two on a remote rack, and replica three beside replica two, balancing fault domains against cross-rack traffic. Writes pipeline client to first DataNode to second to third with acknowledgments reversing, so one network send per hop streams 64KB packets without duplicating client traffic.

Where HDFS locality beats S3 scale and where it loses

S3, a cloud object store holding immutable keys over HTTP, offers serverless infinite scale with per-prefix request limits and multipart commit. HDFS offers data local to executors, append-only writes with atomic rename-to-commit, and a cluster you operate with NameNode failover plus balancer. The trade is operational burden for commit semantics and locality: keep HDFS where Spark or MapReduce commits thousands of part-files per job, move to objects where durability without servers matters more than colocated compute.

Counting NameNode megabytes before ten million tiny files count you

Every file, directory, and block lives in NameNode heap at roughly 150 bytes per inode plus 150 per block. A 1GB file in 128MB blocks costs one inode plus eight block entries, about 1.4KB. That sounds tiny until a Spark job writes ten million 4KB part-files: 40GB of data costing 3GB of master RAM plus collection pauses, for bytes that fit on a laptop.

Heap math (directional, plan with 200 bytes to be safe):
  10M files × 1 block each ≈ 10M × (150 + 150)B ≈ 3 GB heap
  100M blocks (10M files × 10 blocks) ≈ 100M × 150B + inodes ≈ 15–18 GB heap
  10M tiny 4KB files = 40GB of data costing ~3GB of master RAM + GC pain

Fixes that ship:
  bigger blocks (256–512MB) for archival data → fewer entries per GB
  HAR / SequenceFile / CombineFileInputFormat → merge small files before HDFS
  Federated NameNodes (viewFS) or migration to object store for the long tail

Following one 256MB write down its three-node pipeline

Your client writes 256MB with replication 3. It streams to the first DataNode, which streams to the second, which streams to the third, while acknowledgments flow back up. Pipelined 64KB packets mean the client sends once while all three copies form downstream. If the first DataNode fails mid-write, the client re-pipelines around it and the NameNode re-replicates later. Reads invert the path: the NameNode returns replicas sorted nearest-first and the client connects directly.

client → DN1 (rack A) → DN2 (rack B) → DN3 (rack B, different node)
  packets stream DN1→DN2→DN3 while acks return DN3→DN2→DN1→client
  DN1 fails mid-write → client re-pipelines excluding DN1, block re-replicated later
  read path: NameNode returns sorted replica list → client reads nearest DN
  rack-aware rule: 1 local rack, 2nd remote rack, 3rd same-remote-rack (2 racks min)

Committing 500 Spark part-files without publishing half-written output

A Spark job with 500 tasks writes 500 part-files while task 217 fails and retries after its first attempt left a partial behind. On HDFS the fix is staging plus atomic rename: tasks write hidden attempt paths and only winners promote to the final directory, so failed partials stay invisible for later sweeping. Object stores relearned this because their early renames were copy-plus-delete, which is slow and non-atomic, so retried tasks double-committed.

HDFS commit: task → _temporary/attempt_217/part-00217 (hidden staging)
  success → atomic rename to output/part-00217 (O(1) metadata op)
  retry of 217 → new attempt id, old partial never visible, cleaner deletes staging
S3A evolution: directory committer (rename emulation) → partitioned committer
  → magic committer: tasks write unique keys, manifest lists winners on job commit
  no renames at all; commit = writing one small _SUCCESS manifest file

Sizing blocks for Spark parallelism instead of 2006 defaults

The 128MB default tuned MapReduce mapper counts, and Spark inherits the arithmetic because one HDFS block still means roughly one partition. Ten terabytes at 128MB yields 80,000 partitions of healthy parallelism, while 4MB chunks would drown the driver in scheduling. Rejected alternative: shrink to 4MB blocks for finer-grained parallelism. That turns 10TB into 2.5M blocks costing 375MB of NameNode heap at 150 bytes each, plus 2.5M scheduled tasks the driver must track, so the finer split buys scheduling collapse instead of speed. Bigger blocks mean fewer tasks and a happier NameNode, traded against coarser retry units when one large task fails.

Balancing a fleet where new disks idle at 5% while old sit at 85%

Three new DataNodes join while old disks sit at 85% and newcomers idle at 5%. New writes favor empty nodes but old imbalance never self-heals, since HDFS never splits blocks. The balancer moves whole blocks until every node sits within a threshold such as 10% of average, throttled so production reads never notice. Decommission reverses the flow: mark, replicate away, verify empty, then power off. Killing a full node outright triggers an under-replicated storm plus slow reads until recovery.

Balancer: threshold 10% → moves blocks until every DN within 10% of cluster average
  bandwidth default dfs.datanode.balance.bandwidthPerSec = 10MB/s (raise off-peak)
  10TB to move at 100MB/s throttled ≈ 28h; start Friday, not Monday morning
Decommission: hdfs dfsadmin -decommission datanode → NameNode re-replicates its
  blocks elsewhere first; node reports "Decommissioned" only when empty, then power off.
  killing a full DN outright = under-replicated storm + slow reads until recovery.

Federating one overloaded namespace into three independent brains

One namespace with 400M files keeps the NameNode heap in permanent alert while teams refuse shared fate. Federation, meaning splitting the namespace across independent NameNodes, separates analytics, archive, and scratch, each with its own heap and failure domain, unified under one ViewFS mount table, meaning a client-side path map with no extra hop. The price is no cross-namespace renames and no shared transactions. Router-based federation adds a stateful gateway for closer single-cluster behavior at the cost of running the router tier.

ViewFS mount table (client-side, no extra hop):
  /analytics → nn-analytics:8020  (200M files, 30GB heap, SSD journal)
  /archive   → nn-archive:8020    (150M files, erasure-coded, slow disks)
  /scratch   → nn-scratch:8020    (ephemeral, small heap, aggressive expiry)
Router-based federation (RBF): stateful gateway routes + quota across namespaces;
  closer to "one cluster" UX, at the cost of running the router tier itself.

Short-circuiting the local read that loops through two sockets for no reason

A Spark executor sharing a machine with a DataNode still reads executor to socket to DataNode to socket to executor, crossing the kernel twice plus checksums for bytes already in page cache. Short-circuit reads, meaning the client opening the block file directly after one verified file-descriptor handoff, skip the DataNode and stream zero-copy. Same bytes, no loopback, roughly double local throughput on data-local stages where locality finally pays.

Normal: executor → TCP(DataNode) → disk → TCP → executor (2 copies, 2 syscalls per chunk)
Short-circuit: executor opens block file fd → reads page cache directly (0 copies)
  guard: DataNode passes the fd only after verifying the replica + checksums match
  payoff: local reads ~2x throughput: the "data locality" dividend, actually collected.

Recovering the NameNode that never comes back after a power cut

The active NameNode, meaning the master holding the namespace in RAM, dies with edits still in its journal. The naive recovery replays the last checkpoint plus edit log on the standby, which works only if the shared edits actually reached quorum journal nodes before the crash. The working path promotes the standby holding the longest committed edit sequence, replays from its last fsimage, meaning the serialized namespace snapshot, plus rolled edits, then reconciles DataNode block reports that arrive over the next minutes. Blocks written but never journaled appear as orphans the NameNode deletes after leases expire, while under-replicated committed blocks re-replicate from survivors. Checkpoint frequency bounds the replay: hourly checkpoints mean at most an hour of edits to roll, daily ones mean a day.

Crash: active dies at edit 8.2M, last checkpoint fsimage at 8.0M + 200k edits
  promote: standby with edits through 8.19M (quorum-committed) wins over 8.15M peer
  replay: load fsimage 8.0M → roll 190k edits (~minutes) → serve reads, queue writes
  reconcile: DataNodes report blocks over ~10min; orphan uncommitted blocks deleted
    after lease expiry, under-replicated committed blocks re-replicate from survivors.
  rule: checkpoint hourly, journal on quorum; replay time is checkpoint age.

When HDFS still earns its servers

HDFS commits 500 part-files atomically through staging plus rename, and the NameNode replays 190k edits in minutes after a crash. Yet every output lands at once when the whole job finishes. What serves payment-succeeded to fraud, warehouse, and analytics at three different speeds, and replays last month for one team without rerunning anything?