In the previous lesson we learnt how to decide which node stores each key. Now we go one level down and look at the system that physically stores the bytes in such a way that many clients, on many machines, see them as ordinary files. A distributed file system (DFS) is the oldest form of shared network storage, and it is still the one that most resembles what every programmer knows: directories, files, open, read, write. But beneath that familiar interface lie very different decisions: NFS was designed so that an office could share one server's disk; GFS and HDFS so that thousands of cheap machines could hold petabytes of huge files that are written once and read many times; GlusterFS and Ceph so as not to depend on any central node. For Kilometre Zero the question is a concrete one: where to accumulate the millions of events from orders.events and the website's click logs so that Module 5 can process them in bulk. The answer will be HDFS, and we will set it up in the docker-compose.yml of km0/.

1. What a distributed file system is and which transparencies it offers
2. NFS: the client-server model
3. GFS and HDFS: huge files on cheap machines
4. NameNode high availability
5. What HDFS cannot do
6. GlusterFS and Ceph: no centralised metadata
7. Comparison table and use cases
8. HDFS at Kilometre Zero: the data lake
9. Hands-on: HDFS in docker-compose.yml and the WebHDFS API
10. Common Mistakes and Tips
11. Exercises
12. Conclusion

A DFS presents its clients with a hierarchical namespace (directories and files) whose contents are stored on one or more remote servers. What sets it apart from "copying files over the network" is that access is integrated into the operating system, or into an equivalent API, so that programs do not notice (or notice as little as possible) where the data is. That "not noticing" is exactly the concept of transparency we defined in 01-01, and a DFS is the best example for revisiting it:

The delicate point in any DFS is its sharing semantics: what one client sees while another client is writing the same file. On a local system (UNIX semantics) every write is immediately visible to any subsequent read. Over the network, guaranteeing that requires every operation to travel to the server, which kills performance. Each DFS picks a different point between performance and semantics, and that point explains almost all of their differences.

NFS (Network File System, Sun, 1984; now at version 4.2) is the classic DFS: a server exports a directory and the clients mount it in their local tree. Every client operation becomes an RPC call (the ONC RPC we saw in 02-02, with XDR as its serialization) to the server.

Client caches and consistency. Without a cache, every read would be a network round trip. That is why the NFS client caches data and attributes, and this is where the trade-off appears: if client A has a block cached and client B writes that same block on the server, A will read stale data until it revalidates. NFS uses close-to-open semantics: one client's changes are guaranteed to be visible to the others only once the writer calls close and the reader calls open afterwards. In between, the client revalidates the file's attributes every few seconds (actimeo, 3–60 s by default) and discards its cache if the mtime has changed. It is eventual consistency with a window of seconds, and in the vocabulary of 03-01 it does not even offer read-your-writes across different machines. NFSv4 adds delegations: the server can hand a client the exclusive right to a file for as long as nobody else asks for it, which makes it possible to cache safely and to revoke when another client appears.

Limits. NFS has a single server per export: that is the limit on capacity and performance, and the single point of failure (it can be made highly available with an active/passive pair and shared storage, but it does not scale horizontally). Version 4.1 introduced pNFS (parallel NFS), which separates metadata from data and allows reading from several servers, but adoption has been limited.

When it is still a valid choice. Very often. Sharing a development team's /home directory, giving several instances of an application access to the same configuration files or to an exchange directory, mounting ReadWriteMany volumes in Kubernetes for moderate workloads: NFS is simple, it is in every kernel, and up to a few terabytes and hundreds of clients it just works. Kilometre Zero ruled it out as the store for product photos for two reasons we will see in 04-03: it wants real replication between nodes and a direct HTTP API for the website, not a mount.

In 2003 Google published the design of its Google File System (GFS), and Yahoo reimplemented it as open source under the name HDFS (Hadoop Distributed File System). Its premises break with NFS:

• Files are huge (gigabytes or terabytes), and there are a few million of them, not billions.
• They are written once (or only appended to) and read many times, almost always sequentially and in full.
• The hardware is cheap and fails constantly: with 1,000 disks, one of them dies every day. Fault tolerance is the norm, not the exception.
• What matters is aggregate bandwidth (reading a terabyte in a minute from a hundred machines), not the latency of a small read.

Architecture. HDFS has two kinds of node:

The key concepts:

• Large blocks: 128 MB by default (compared with the 4 KB of a local file system). A 1 GB file is 8 blocks. Large blocks reduce the amount of metadata the NameNode keeps in memory and make every read a long sequential transfer, which is what disks are good at. A file smaller than a block takes up only its actual size (it does not waste 128 MB), but it does consume a full metadata entry.
• Replication factor: each block is copied to dfs.replication DataNodes, 3 by default. Replication is per block, not per file, and the NameNode keeps watch over it: if a DataNode stops sending heartbeats (10 minutes by default), all its blocks become "under-replicated" and the NameNode orders them to be copied from the surviving replicas to other nodes.
• Rack awareness: the NameNode knows which rack each DataNode is in. With 3 replicas, it places the first on the client's node (if it is a DataNode), the second on a node in another rack and the third on a different node in the same rack as the second. That way the failure of an entire rack (a switch) loses no blocks, and only one of the three copies crosses between racks (inter-rack bandwidth being the scarce resource).
• Write-once / append: a file is created, written (by a single writer) and closed; after that it is immutable, except for appending to the end (append). There are no writes in the middle of a file. This simplifies consistency enormously: there are never two concurrent writers on the same bytes, and the replicas of a closed block are identical for ever.

Write and read flow. The essential point is that data never passes through the NameNode; only metadata does:

On a write, the client asks the NameNode where to put the block and sends the data to the first DataNode, which forwards it to the second, which forwards it to the third (the replication pipeline): the client sends the data only once, and replication uses bandwidth between DataNodes, not the client's. The block is considered written once all three have acknowledged: this is synchronous replication in the sense of 03-04, and it is why HDFS is CP (a write is not accepted if it cannot reach the minimum number of replicas, dfs.namenode.replication.min, which defaults to 1). On a read, the client obtains the list of blocks with their locations and reads each block directly from the nearest DataNode; if one fails, it moves on to the next in the list. Every block carries checksums (CRC32 every 512 bytes) which the client verifies: a corrupt block is discarded, read from another replica, and reported to the NameNode so that it can be re-replicated.

In the original design the NameNode was a single point of failure: if it went down, the whole cluster became inaccessible, even though the data was still intact on the DataNodes. Since Hadoop 2 there has been a high availability (HA) configuration, which applies what we already know from 03-03 and 03-04:

• Two (or more) NameNodes: one active and one on standby. Both hold the namespace in memory.
• The active node's change log (the edit log) is written to a quorum of JournalNodes (normally 3): a metadata operation is committed once a majority has persisted it. The standby NameNode reads that log continuously and applies the changes, so its copy is only a few milliseconds behind. This quorum is a direct application of the W > N/2 from 03-04.
• DataNodes send heartbeats and block reports to both NameNodes, so that the standby knows where every block is and can take over without rebuilding that information.
• ZooKeeper (03-03) decides who is active: each NameNode runs a ZKFailoverController process that maintains an ephemeral node in ZooKeeper; if the active one stops renewing it, the standby acquires the lock and promotes itself.
• Fencing: before promoting itself, the new active node makes sure the old one can no longer write to the JournalNodes (which only accept writes from the NameNode with the highest epoch, just like Raft terms), and optionally kills it over SSH. This is the defence against the split-brain we already discussed in 03-04.

With HA, a failure of the active NameNode is resolved in tens of seconds with no intervention. Without HA (as in our hands-on docker-compose), a SecondaryNameNode merely compacts the edit log; it is not a backup and it cannot take over, a misunderstanding so common that it deserves underlining.

Its premises are its limits, and it is worth being clear about them before putting something into HDFS that does not fit:

• Lots of small files. Every file, directory and block takes up about 150 bytes of NameNode memory. A hundred million 10 KB files means 15 GB of heap and 1 TB of data that, in 128 MB blocks, would have been 8,000 entries. On top of that, every read of a small file pays for a call to the NameNode and a connection to a DataNode just to read a few KB. Product photos (thousands of 200 KB files) are a bad fit for HDFS; that is why they go to an object store (04-03). If small files do have to go into HDFS, they are grouped together (SequenceFile files, HAR or, better still, daily partitions consisting of a single large file, as we will do with the events).
• Low-latency random access. Reading one specific record means locating the block, opening a connection and reading at least one packet; tens of milliseconds. HDFS is not a database: HBase was built on top of it precisely to provide random access, managing its own indexes and large files.
• Concurrent writers and modifications in the middle of a file. A file has a single writer and only supports append. A log that many services write to at once has to go through Kafka (02-04) first and be flushed to HDFS in batches.
• Full POSIX semantics. There is no mmap, no locks and no partial writes; "mounting" it via the NFS Gateway or FUSE is a compatibility layer with limitations.

The NameNode, even with HA, is a limit: the whole namespace has to fit in one machine's memory and every metadata operation goes through it. A second family of DFSs does away with the metadata server and locates data by computation, using the same ideas as the consistent hashing of 04-01.

GlusterFS pools directories exported by several servers (bricks) into a volume. There is no metadata server: the location of each file is computed with a hash of its name over a range assigned to each brick (elastic hashing), and the client, which knows the volume configuration, talks directly to the right brick. Volumes can be distributed (each file on one brick), replicated (each file on N bricks) or dispersed (with erasure coding, which we will see in 04-03), and these can be combined. It is simple, it mounts like an ordinary file system (FUSE or NFS) and it scales well for medium and large files; its weak points are directory operations (an ls has to ask every brick) and self-healing after failures.

Ceph is more ambitious: a distributed object store (RADOS) on top of which three interfaces are built: block (RBD, virtual disks for virtual machines and Kubernetes), object (RGW, S3-compatible, 04-03) and file (CephFS). Its components:

• OSD (Object Storage Daemon): one process per disk, which stores objects and replicates peer-to-peer with other OSDs.
• Monitors (MON): a small quorum (Paxos, 03-03) that maintains the cluster map (which OSDs exist and their state). They store neither data nor file metadata.
• CRUSH (Controlled Replication Under Scalable Hashing): the algorithm that, given an object's name and the cluster map, computes which OSDs hold its replicas. It is a direct relative of the consistent hashing of 04-01, with one important difference: CRUSH understands the topology (disk → server → rack → room) and a set of rules ("three replicas in three different racks"), so that replicas are not only spread evenly but also respect failure domains. Any client with the map computes the location without asking anyone: there is no NameNode, and adding an OSD moves only the proportional fraction of objects, just as in the ring.
• MDS (Metadata Server): for CephFS only, it manages the directory tree; several can be active, dynamically sharing out subtrees between them.

Ceph is the storage engine behind many private clouds (OpenStack, Proxmox, Kubernetes with Rook). It is also considerably more complex to operate than HDFS or GlusterFS.

In 01-06 we established that analytics would process historical data in batch and recent data as a stream. Historical data needs somewhere to accumulate for years: cheap per terabyte, fault-tolerant and optimised for Module 5 to read it in full and in parallel. That place is the data lake on HDFS, with two sources:

• The events from the orders.events topic (order.created, stock.reserved, payment.confirmed, … with the event_id/type/version/timestamp_ms/source/data envelope from 02-05). Kafka retains events for a few days; an analytics consumer flushes them to HDFS in batches, one file per day and type: /km0/events/2026-09-14/orders.jsonl. Each line is one event in JSON (the JSON Lines format). With 40,000 orders and about 6 events per order, a campaign day comes to roughly 250,000 events and about 150 MB: one or two blocks, an ideal size for HDFS.
• The website's click logs (page viewed, product viewed, added to basket), which the web server writes to files rotated hourly and which a process uploads to /km0/clicks/2026-09-14/hour=13/web-01.jsonl.

Organising by date directories (date=2026-09-14/) is no accident: it is the range partitioning of 04-01 applied to files, and it will let Module 5 process "just Grape Harvest Week" without reading the rest of the lake. The events are immutable (a perfect fit for write-once), access is sequential and in bulk, and nobody needs to read "event 123" with low latency: that is what the orders database (04-04) is for. This is the division that the module will end up with: HDFS for what is massive and cold, the database for what is operational, objects for the files the website serves.

We add a NameNode and two DataNodes to the docker-compose.yml of km0/, using the official apache/hadoop image. The image is configured with environment variables whose names mirror Hadoop's XML files (CORE-SITE.XML_fs.defaultFS is equivalent to the fs.defaultFS property in core-site.xml):

Points to understand:

• dfs.replication: 2 because there are only two DataNodes; with the default value (3) every block would be permanently "under-replicated" and fsck would flag it as a warning.
• ENSURE_NAMENODE_DIR makes the container run hdfs namenode -format if the directory is empty. Formatting a NameNode that holds data wipes the namespace (the blocks are left orphaned on the DataNodes), so it lives on a persistent volume.
• Port 9870 is the NameNode's web console (http://localhost:9870), where you can see the live DataNodes, the capacity and the file browser.

We start it up and try it out with the CLI, run inside the NameNode container:

hdfs fsck is the tool for seeing the anatomy of a file: blocks, replicas and which DataNodes they are on:

Abridged output:

We see a single block (the file is smaller than 128 MB), with two live replicas at two different addresses, and a single rack (we have not configured the topology). An instructive experiment: docker compose stop datanode-2, wait about 10 minutes (or lower dfs.namenode.heartbeat.recheck-interval to speed things up) and repeat the fsck: the block shows up as under-replicated with Live_repl=1, and the -cat still works thanks to the replica on datanode-1. When datanode-2 is started again, it goes back to Live_repl=2.

WebHDFS from Python. HDFS exposes its full API over HTTP (WebHDFS), which saves installing a Java client in analytics. Operations that touch data use a two-step redirect that mirrors the flow in section 3: you ask the NameNode to CREATE (without sending any data), the NameNode replies with a 307 carrying the URL of a DataNode, and the client sends the data to that DataNode. The script services/analytics/upload_events_hdfs.py uploads the day's events file:

Explanation:

• _url builds the WebHDFS URL: the HDFS path goes in the URL path and the operation in op. user.name identifies the user (without Kerberos, HDFS simply trusts the name: this is why authentication, the subject of 06-01, is enabled in production).
• create performs the two-step dance with allow_redirects=False: we want to see the 307 and read the Location header, which points to http://datanode-1:9864/webhdfs/v1/.... From outside Docker that name does not resolve; add 127.0.0.1 datanode-1 datanode-2 to /etc/hosts (and port 9864 only works for datanode-1, which is why datanode-2 publishes on 9865: in a real environment the client would be on the same network as the DataNodes, and this nuisance goes away).
• append is the same pattern with POST. Only one client can have the file open for writing; a second, concurrent APPEND gets an AlreadyBeingCreatedException error, which is the single-writer semantics of section 3 showing through HTTP.
• list_dir returns, among other things, replication and blockSize, which confirm what fsck showed.

With the analytics Kafka consumer accumulating the day's events in events/2026-09-14/orders.jsonl and this script run every hour (or when the day closes), the data lake grows by one directory per day, ready for Module 5.

• Treating HDFS as a network drive. It is not POSIX, it supports neither random writes nor concurrent writers, and every small file costs the NameNode memory. Always group: one large file per day and source.
• Believing the SecondaryNameNode is a backup. It only compacts the edit log. Without JournalNodes and ZooKeeper there is no high availability, and without copies of the metadata directory (dfs.namenode.name.dir on several disks) a hardware failure can leave the blocks orphaned and unrecoverable.
• A replication factor greater than the number of DataNodes. Everything stays under-replicated for ever and the NameNode retries endlessly. Set dfs.replication to suit the cluster.
• Ignoring rack topology. Without net.topology.script.file.name, HDFS believes everything is in one rack and may put all three replicas behind the same switch.
• Relying on the NFS client cache to coordinate processes on different machines. Close-to-open semantics do not guarantee that machine B sees what A has just written until A closes and B opens. If you need coordination, use a distributed lock (etcd, 03-03) or a queue, not the file system.
• Using WebHDFS without planning for DataNode name resolution. The redirect returns the DataNode's hostname; the client must be able to resolve it and reach its port. It is the number one mistake when using WebHDFS from outside the cluster.
• Choosing Ceph "because it scales further" for a 10 TB problem. Its operational complexity is real. For a batch data lake, HDFS (or an object store, 04-03) is simpler; for moderate shared volumes, NFS is still the right answer.

Exercise 1. Every day Kilometre Zero generates 250,000 order events (about 150 MB in JSON Lines) and 12 million click lines (about 4 GB, in hourly files from each of 3 web servers). An engineer proposes uploading every click file to HDFS as is (3 servers × 24 hours = 72 files/day of ~55 MB) and, in addition, one file per order for the events (40,000 files/day). Work out for one year: (a) the number of files and blocks for each option, (b) the approximate NameNode memory (150 bytes per file and per block), and (c) propose a better layout.

Exercise 2. With the hands-on docker-compose, explain what happens step by step, in terms of the NameNode, the DataNodes and the client, when you run hdfs dfs -cat /km0/events/2026-09-14/orders.jsonl while datanode-1 is stopped (without waiting for the 10-minute detection period). Which WebHDFS header would make the Python script fail in the same situation, and how would you make it robust?

Exercise 3. Write a function check_replication(path) that uses the WebHDFS GETFILEBLOCKLOCATIONS operation (op=GETFILEBLOCKLOCATIONS) to list, block by block, the hosts that hold a replica, and returns the list of blocks with fewer replicas than configured. Explain how this check resembles what the NameNode itself does and why, on a large cluster, it is not a good idea to run it over the whole lake every minute.

Solution 1:

(a) The engineer's proposal, one year (365 days): clicks: 72 × 365 = 26,280 files, each 55 MB file fits in 1 block → 26,280 blocks. Events: 40,000 × 365 = 14.6 million files of about 4 KB, 1 block each → 14.6 million blocks. Total ≈ 14.63 million files and as many blocks. Better layout: one click file per day (4 GB → 32 blocks of 128 MB): 365 files and 11,680 blocks; one events file per day (150 MB → 2 blocks): 365 files and 730 blocks. Total: 730 files and 12,410 blocks.

(b) NameNode memory: the engineer's proposal ≈ (14.63 M files + 14.63 M blocks) × 150 B ≈ 4.4 GB of heap per year for this data alone (and growing; each replica also adds an entry to the block map). Better layout ≈ (730 + 12,410) × 150 B ≈ 2 MB. Three orders of magnitude apart for the same volume of data (about 1.5 TB a year).

(c) Consolidate by day and source (/km0/clicks/2026-09-14/clicks.jsonl, /km0/events/2026-09-14/orders.jsonl), accumulating in the Kafka consumer and uploading in batches with append every hour; optionally compress with a splittable format (Parquet or Avro, which Module 5 will read better than JSON) and apply a lifecycle that archives older years with a replication factor of 2 or with erasure coding (Hadoop 3 supports it).

Solution 2:

The client asks the NameNode for getBlockLocations; as the 10 minutes have not yet passed, the NameNode still believes datanode-1 is alive and returns the block with two locations, [datanode-1, datanode-2] (or the other way round, depending on the computed proximity). The client tries to connect to the first one; if that is datanode-1, the connection fails (refused, or timing out after dfs.client.socket-timeout, 60 s by default, which may be noticeable as a pause), it marks that DataNode as "dead for this client" and moves on to the next in the list, datanode-2, which serves the block. The -cat works, perhaps after an initial delay. Nothing is re-replicated until the NameNode detects the outage. In WebHDFS, the problem is the Location header of the redirect: the NameNode may redirect an OPEN or CREATE to datanode-1, and the second step will fail with a connection error. To make it robust: catch requests.ConnectionError in the second step and retry the operation from the first step (the NameNode will pick another DataNode on the retry, especially if you pass it the excludedatanodes=datanode-1 parameter on reads), with a small number of retries and an explicit timeout in requests (timeout=(5, 60)), which is exactly the retry-with-timeout pattern that 07-04 will formalise.

Solution 3:

GETFILESTATUS gives the file's replication factor and GETFILEBLOCKLOCATIONS (available since Hadoop 2.8/3.x) returns, for each block, the hosts holding a replica. The NameNode continuously does something equivalent, but from the other side: it cross-checks the block reports each DataNode sends it against the expected factor and maintains a queue of under-replicated blocks, which it works through under a bandwidth limit. Running the external check over the whole lake every minute is a bad idea because every metadata call competes with real operations for the same NameNode (a single namespace write thread, with a global lock) and because with tens of thousands of files it means tens of thousands of HTTP requests; the sensible approach is to query the aggregate metrics the NameNode already computes (UnderReplicatedBlocks in its JMX or in hdfs dfsadmin -report) and keep fsck or this function for investigating specific files, a topic we will return to in 07-01.

A distributed file system offers the most familiar storage interface (directories and files) on top of many nodes, and its character is determined by the sharing semantics it chooses. NFS keeps a single server and close-to-open consistency backed by client caches; it is still the right answer for moderate shared directories. GFS and HDFS give up POSIX to achieve what NFS cannot: petabytes on cheap machines, with a NameNode that keeps the metadata in memory, DataNodes that serve 128 MB blocks replicated three times with rack awareness, a synchronous write pipeline, a single writer per file, and high availability based on a quorum of JournalNodes and a ZooKeeper election, which are the tools of 03-03 and 03-04 put to work. That design fails, by construction, with lots of small files and with random access. GlusterFS and Ceph do away with the metadata server and locate data by computation, with CRUSH as a relative of the consistent hashing of 04-01 that also respects failure domains. At Kilometre Zero, HDFS is the data lake: one file per day for the orders.events events and the click logs, which we have set up with apache/hadoop in docker-compose.yml, inspected with hdfs dfs and hdfs fsck, and fed with upload_events_hdfs.py through WebHDFS and its two-step redirect.

The product photos have been left out on purpose: thousands of small files that the website must serve directly over HTTP, with metadata, versions and a durability that does not depend on a NameNode. For that there is a different model, with no directories and no append, and with an API that has become the de facto standard of the cloud: object storage, which we will look at next with MinIO and boto3.