Distributed File System (HDFS / GFS)

Problem statement (interviewer prompt)

Design a GFS/HDFS-style distributed file system for analytics workloads. Files are large (GB-TB), writes are append-only, reads are bandwidth-bound, and the namespace must survive single-machine failures. Cover the master/datanode split, chunk placement, and recovery.

flowchart LR
  C([Client])
  NN[NameNode<br/>metadata + namespace]
  DN1[(DataNode 1)]
  DN2[(DataNode 2)]
  DN3[(DataNode 3)]
  C -->|open file| NN
  NN -->|chunk locations| C
  C -->|read/write| DN1
  C --> DN2
  C --> DN3
  DN1 <-. replicate .-> DN2

    classDef client fill:#dbeafe,stroke:#1e40af,stroke-width:1px,color:#0f172a;
    classDef edge fill:#cffafe,stroke:#0e7490,stroke-width:1px,color:#0f172a;
    classDef service fill:#fef3c7,stroke:#92400e,stroke-width:1px,color:#0f172a;
    classDef datastore fill:#fee2e2,stroke:#991b1b,stroke-width:1px,color:#0f172a;
    classDef cache fill:#fed7aa,stroke:#9a3412,stroke-width:1px,color:#0f172a;
    classDef queue fill:#ede9fe,stroke:#5b21b6,stroke-width:1px,color:#0f172a;
    classDef compute fill:#d1fae5,stroke:#065f46,stroke-width:1px,color:#0f172a;
    classDef storage fill:#e5e7eb,stroke:#374151,stroke-width:1px,color:#0f172a;
    classDef external fill:#fce7f3,stroke:#9d174d,stroke-width:1px,color:#0f172a;
    classDef obs fill:#f3e8ff,stroke:#6b21a8,stroke-width:1px,color:#0f172a;
    class C client;
    class NN service;
    class DN1,DN2,DN3 datastore;

flowchart TB
  subgraph Client[Client]
    APP([App / MapReduce / Spark])
    LIB([DFS Client library])
  end

  subgraph Master[Master Tier]
    NN[NameNode / Master<br/>namespace + chunk map]
    SNN[Standby NameNode<br/>HA via Raft / Quorum Journal]
    LEASE[Lease manager]
    BR([Block report aggregator])
    BAL[Balancer]
    QUOTA[Quota / ACL]
  end

  subgraph Storage[Chunk Servers / DataNodes]
    direction LR
    DN1[(DataNode 1<br/>chunks 64 MB-256 MB)]
    DN2[(DataNode 2)]
    DN3[(DataNode 3)]
    DNN[(DataNode N)]
  end

  subgraph Pipeline[Write Pipeline]
    PRIMARY[Primary replica<br/>holds lease]
    SEC1[Secondary 1]
    SEC2[Secondary 2]
    ACK[Daisy-chain ack]
  end

  subgraph EC[Erasure Coding optional]
    RS[Reed-Solomon 10+4 / 6+3]
    SAVE[~40% storage save vs 3x replication]
    RECONS[Reconstruction reads on degraded]
  end

  subgraph Recovery
    HB[Heartbeats every few seconds]
    DETECT[Failure detector]
    REPL[Re-replication]
    SCRUB[Checksum scrubber]
  end

  subgraph App[Workloads]
    MR([MapReduce / Spark / Flink])
    LOG[Log archive]
    ML[ML training shards]
    BQ[BigQuery / Hive tables]
  end

  APP --> LIB
  LIB -->|metadata RPC| NN
  NN -. log + checkpoint .-> SNN
  LIB -->|read| DN1
  LIB -->|read| DN2
  LIB -->|read| DN3
  LIB -->|write| PRIMARY --> SEC1 --> SEC2 --> ACK --> PRIMARY
  HB --> NN
  BR --> NN
  NN -.replicate cmd.-> REPL
  REPL --> Storage
  EC --- Storage
  App --- LIB

    classDef client fill:#dbeafe,stroke:#1e40af,stroke-width:1px,color:#0f172a;
    classDef edge fill:#cffafe,stroke:#0e7490,stroke-width:1px,color:#0f172a;
    classDef service fill:#fef3c7,stroke:#92400e,stroke-width:1px,color:#0f172a;
    classDef datastore fill:#fee2e2,stroke:#991b1b,stroke-width:1px,color:#0f172a;
    classDef cache fill:#fed7aa,stroke:#9a3412,stroke-width:1px,color:#0f172a;
    classDef queue fill:#ede9fe,stroke:#5b21b6,stroke-width:1px,color:#0f172a;
    classDef compute fill:#d1fae5,stroke:#065f46,stroke-width:1px,color:#0f172a;
    classDef storage fill:#e5e7eb,stroke:#374151,stroke-width:1px,color:#0f172a;
    classDef external fill:#fce7f3,stroke:#9d174d,stroke-width:1px,color:#0f172a;
    classDef obs fill:#f3e8ff,stroke:#6b21a8,stroke-width:1px,color:#0f172a;
    class LIB client;
    class NN,SNN,LEASE,BAL,QUOTA,PRIMARY,SEC1,SEC2,ACK,RS,SAVE,RECONS,HB,DETECT,REPL,SCRUB,LOG,ML service;
    class DN1,DN2,DN3,DNN,BQ datastore;
    class APP,BR,MR compute;

Design choices

• Append-only writes (GFS/HDFS) - random writes not supported, simplifies replication.
• Large chunk size (64-256 MB) - fewer metadata entries, sequential IO friendly.
• Single master with HA standby - simple to reason about, but bottleneck → modern Colossus / HDFS Federation use multiple namespaces.
• Replication 3 by default; EC for cold data.

Read path

1. Client asks master for chunk locations of file offset.
2. Client streams directly from a chosen DataNode (often rack-local).
3. Checksums verified per block.

Write path

1. Client requests new chunk → master assigns 3 DataNodes, picks primary (holds lease).
2. Client pushes data to closest replica → it forwards down chain (daisy-chain saves bandwidth).
3. Primary orders mutations; secondaries apply; all ack back.
4. Master updates metadata only on chunk completion.

Master scaling

• Federation: namespace split across multiple NameNodes.
• Colossus: separates metadata (Spanner-backed) from old single-master design.

Glossary & fundamentals

Concepts referenced in this design. Each row links to its canonical page; the tag column shows whether it is a high-level (HLD) or low-level (LLD) concept.

Tag	Concept	What it is	Page
HLD	Raft / Paxos consensus	replicated state machine via majority quorum	consensus-raft-paxos
HLD	Leader/follower replication	sync/semi-sync/async replication, failover	replication-leader-follower
HLD	Batch & stream processing	Lambda vs Kappa, watermarks, windows	batch-stream-processing

Quick reference

Goal

A single namespace usable by data-processing frameworks, with bandwidth-saturating sequential reads/writes and rack-aware placement.

Properties

• Large files (GB-TB).
• Append + truncate workloads (logs, MR output).
• Tolerate disk/node/rack failures.
• Horizontal capacity scaling by adding DataNodes.

Capacity

• HDFS at Yahoo / FB scaled to ~100 PB per cluster, billions of files.
• Colossus underpins essentially all Google storage.

Schema (metadata)

• inode(id, type, parent_id, name, mtime, perm)
• chunks(file_id, offset, chunk_id, locations[])
• chunk_servers(id, host, capacity, last_heartbeat)

Trade-offs

• Append-only is great for analytics; bad for OLTP-like workloads → split storage by purpose.
• Single master = simpler but bottleneck → federation / Colossus-style separation.
• Large blocks = great bandwidth, bad for many tiny files (NameNode RAM pressure).
• 3× replication vs EC: replication faster recovery + reads; EC saves cost on cold data.

Refs

• "The Google File System" (SOSP '03).
• HDFS Architecture guide (Apache).
• Colossus blog posts (Google).
• "Tectonic" Facebook DFS paper.

FAQ

What does the NameNode store in HDFS or GFS?

The NameNode keeps the file system namespace, directory tree, and a map from each file's chunks to the DataNodes that hold them. DataNodes report block locations via periodic heartbeats.

Why does HDFS default to a replication factor of three?

Three replicas survive two simultaneous machine failures with high probability while keeping storage overhead at 3x. Replicas are placed across racks to tolerate rack level outages.

How are large files split in a distributed file system?

Files are split into fixed size chunks, typically 64 MB or 128 MB, and each chunk is replicated independently. Large chunk sizes reduce NameNode metadata pressure and favor sequential reads.

How does HDFS recover from a DataNode failure?

When the NameNode stops receiving heartbeats it marks the DataNode dead. Under replicated blocks are then re-replicated to other nodes until the replication factor is satisfied again.

Why is HDFS optimized for append only writes?

Analytics workloads bulk write large files and read them sequentially. Append only semantics avoid expensive coordination for random writes and let chunks be replicated with a simple pipelined protocol.

Video walkthrough

Distributed Storage Systems: HDFS & GFS Explained : via System Design Walkthrough