Problem RestatementProblem
Design a distributed file system (asked at Databricks) that stores very large files (GBs to TBs) across thousands of cheap machines, like Google File System or HDFS. It offers a normal hierarchy (/logs/2026/09/19/part-001) and operations like create, read, append, delete and list. Disks and machines fail every day, so the system must keep data safe and available automatically, and give high throughput for big sequential reads and writes (batch analytics).
RequirementsRequirements
- Namespace: directories and files with permissions.
- Large sequential reads and writes, and appends (random overwrites are rare).
- Survive disk, machine and rack failures without data loss.
- Petabytes of data, thousands of clients reading in parallel.
ArchitectureArchitecture
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
C["Client library"] -->|"1. where are chunks of file X?"| M["Metadata master (NameNode)"]
M -->|"2. chunk IDs + server locations"| C
C -->|"3. read/write data directly"| CS1["Chunk server - rack 1"]
C --> CS2["Chunk server - rack 2"]
C --> CS3["Chunk server - rack 3"]
CS1 -->|"heartbeats + chunk reports"| M
CS2 --> M
CS3 --> M
M --> LOG[("Metadata edit log + snapshot")]
SB["Standby master"] --> LOG- Separate metadata from data: one master (with a standby) holds the namespace (the tree of files) and the mapping file → chunks → chunk servers, all in memory for speed. Clients get locations from it, then move data directly with chunk servers, so the master never becomes a data bottleneck.
- Large chunks (64–128 MB): fewer chunks to track (less metadata) and efficient sequential I/O.
- 3 replicas per chunk, placed rack-aware (e.g., one in the local rack, two in another rack), so a rack failure doesn't lose all copies.
Read Path
- The client asks the master for the chunk locations for a byte range (and caches the answer).
- It reads from the closest replica (same rack if possible) and verifies checksums on each block. On a mismatch, it reads another replica and reports the corruption.
Deep Dive — Writing one record to three replicasDeep dive
A client appends to a file whose chunks live on three chunkservers. All three must end up with the same bytes in the same order, over a network that drops things.
The client writes to each replica itself
The client sends the data to A, then B, then C, and reports success when all three acknowledge.
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
C["Client"] --> A["Replica A"]
C --> B["Replica B"]
C --> D["Replica C"]
C --> UP["The client's uplink carries the data three times"]
W2["A second client appends concurrently"] --> ORDER["A applies X then Y, C applies Y then X"]
ORDER --> DIV["Replicas of the same chunk now differ"]Two independent problems. The client's bandwidth is multiplied by the replication factor, and — worse — nothing decides the order of concurrent mutations, so replicas of the same chunk diverge and there is no way to say which is correct.
Pipeline the data, and let one replica decide the order
Push the bytes along a chain — client → A → B → C — so each link carries the data once and the client's uplink carries it once. Then designate one replica as primary: it assigns a serial order to mutations and instructs the others to apply them in that order.
Order is now well defined and bandwidth is linear. The question left open is who the primary is, and what happens when it stops responding. Without an answer, a partitioned primary and a newly chosen one can both be issuing orderings.
A master-granted lease, and appends that tolerate retries
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
CL["Client"] -->|"where do I write?"| M["Master"]
M -->|"3 replicas + lease to the primary"| CL
CL -->|"push bytes"| A["Replica A"]
A --> B["Replica B"]
B --> D["Replica C"]
CL -->|"commit"| P["Primary - assigns the mutation order"]
P --> A
P --> B
P --> D
P -->|"a replica failed"| RETRY["Client retries - the record may appear twice"]
RETRY --> APPDEDUP["Application deduplicates by record id"]- The lease is time-bounded. A primary holds the right to order mutations for a fixed interval, so a partitioned primary's authority expires on its own and the master can safely grant the lease elsewhere. There is never a moment with two live orderings.
- Data flow is separated from control flow. Bytes travel the pipeline at full link speed; only the small commit message goes to the primary. This is why the design scales with chunk size.
- Appends are at-least-once. A retry after a partial failure can leave a duplicate record, and the file system does not hide that. Applications deduplicate by record id — the cost of a simple, fast append path.
HDFS makes the opposite trade: a single writer per file, which removes concurrent-append ordering entirely at the price of the multi-writer append GFS supports. Naming that difference is usually what the interviewer is after — the two systems solved the same problem with different assumptions about the workload.
Failures and Healing
- Chunk servers send heartbeats and chunk lists to the master. If a server is silent for a while, the master marks it dead and schedules re-replication of its chunks from surviving copies, prioritizing chunks with only 1 copy left.
- A background scrubber verifies checksums, and corrupted replicas are replaced.
- Balancer: moves chunks to even out disk usage.
Master Scalability and HA
- Metadata durability: every namespace change is written to an edit log (replicated, e.g., to a quorum of journal nodes) before being applied, with periodic snapshots (checkpoints).
- High availability: a hot standby master replays the same edit log and takes over on failure (with fencing so the old one can't keep writing).
- Scale limits: all metadata in one master's RAM (~150 bytes per file or chunk) limits the file count. Fixes: federation (several masters, each owning part of the namespace), or a distributed metadata store. Avoid millions of tiny files: pack them into bigger container files.
- Cheaper durability: erasure coding (e.g., 6 data + 3 parity) for cold data instead of 3 full copies, which cuts storage from 3x to 1.5x.
Wrap-UpWrap-up
Split files into large chunks stored with rack-aware 3-way replication on many chunk servers, and keep the namespace and chunk map in memory on a metadata master backed by a replicated edit log and a hot standby. Clients get locations from the master but read and write data directly, writes use pipelines, a leased primary for ordering and checksums. Heartbeats drive automatic re-replication. Scale metadata with federation, and cut cost with erasure coding for cold data.