Google described its file system in a paper in 2003 and MapReduce in another in 2004. Doug Cutting and Mike Cafarella read both and rebuilt the two ideas inside Nutch, their open-source web-search project; in January 2006 that code was carved out as a project of its own — named after the toy elephant of Cutting's son.
Hadoop and HDFS: blocks, replicas, one very important server, YARN
After this lesson you will draw HDFS from memory — the NameNode, the DataNodes, blocks with three replicas on two racks — follow a read and a write message by message, say what happens when a DataNode or the NameNode dies, and explain what YARN split out of the JobTracker.
A hospital heating management system has logged every sensor reading for ten years: one file of 1 TB. Copy it into a Hadoop cluster and the disks of the cluster fill up by 3 TB. That is not a bug. It is the default.
One machine of the cluster knows where every piece of the file lives. Yet when a hundred analysts read the whole file at the same time, not one byte of it passes through that machine.
Two questions for the end of the lesson. Why pay for every byte three times? And how can the most important server of the cluster be the one that never touches the data — and why is exactly that what lets the cluster grow to thousands of machines?
Hadoop: cheap machines that are expected to fail
In lesson 30 MapReduce cut a job into map and reduce tasks spread over many machines. Hadoop is the system that runs it. The slide calls it a "Software platform to easily process vast amounts of data" that includes "MapReduce – offline computing engine", "HDFS – Hadoop distributed file system" and "Hbase – online data access" (p. 835). Offline means batch: you submit a job and collect the result later. HBase, a separate Apache project that adds random access on top of HDFS, belongs to lesson 32.
"Hadoop implements Google's MapReduce, using HDFS", and its "target is to run on clusters of the order of thousands of nodes" (p. 836); HDFS "is a file system written in Java based on the Google's GFS" (p. 846). Fig. 1 redraws the picture of p. 836, how the two halves fit: "MapReduce divides applications into many small blocks of work"; HDFS cuts a file into blocks and copies each block to several machines; each map task runs on a machine that already holds its block — "MapReduce can then process the data where it is located."
Four features, two questions
The four features of p. 835 come in pairs. Scalable — "It can reliably store and process petabytes" — and Economical — "clusters of commonly available computers (in thousands)" — answer the question "how do you afford petabytes?": with many ordinary machines instead of one big one, the scaling out of lesson 3. Efficient — "it can process it in parallel on the nodes where the data is located" — and Reliable — "It automatically maintains multiple copies of data and automatically redeploys computing tasks based on failures" — answer "how do you make thousands of ordinary machines usable?".
Six assumptions, six design decisions
"Hadoop: Assumptions" (p. 837) is the key to the rest of this lesson: every sentence forces a decision.
| Assumption (p. 837) | Design decision |
|---|---|
| "Hardware will fail." | three replicas of every block, heartbeats, re-replication, checksums (section 3) |
| "Processing will be run in batches: High throughput as opposed to low latency." | whole blocks read front to back as streams; random reads are not what HDFS is built for (p. 847) |
| "A typical file in HDFS is gigabytes to terabytes in size." | huge blocks (64 MB on the slides); millions of files, not billions |
| "It should provide high aggregate data bandwidth and scale to hundreds of nodes in a single cluster." | clients read and write the DataNodes directly, never through the master (section 4) |
| "Applications need a write-once-read-many access model." | no update in place, only appends: the copies of a block never disagree |
| "Moving Computation is Cheaper than Moving Data." | block locations are exposed, so each map task runs where its block is (Fig. 1) |
Why assume failure? With thousands of commodity machines something is always broken, so failure is the normal state and recovery must be automatic. Why is moving computation cheaper? A map task is a few kilobytes of code, its input block 64 MB, and the network is the scarce resource of a cluster.
RDBMS vs. Hadoop
| RDBMS | Hadoop | |
|---|---|---|
| Data size | Gigabytes | Petabytes |
| Access | Interactive & Batch | Batch |
| Updates | Read & write many times | Write once, read many times |
| Integrity | High | Low |
| Scaling | Non Linear | Linear |
| Data representation | Structured | Unstructured, semi-structured |
Two rows need care. "Integrity: Low" does not mean that Hadoop loses data — HDFS checksums every block (section 3). It means Hadoop enforces no schema, keys, constraints or transactions: integrity is the application's job. "Scaling: Linear" means that twice the machines give about twice the throughput, because every machine stores and processes its own share.
Hadoop assumes failure and moves the program to the data; every design decision of HDFS follows from the six assumptions.
HDFS: blocks, replicas and the NameNode's two maps
HDFS is "Responsible for storing data on the cluster": "Data files are split into blocks and distributed across the nodes in the cluster", and "Each block is replicated multiple times" (p. 845). "Different blocks from the same file will be stored on different machines" (p. 848), and "Default replication is 3-fold" (p. 849).
Check yourself: which nodes of Fig. 2 can fail together without losing data?
Any two: every block has three copies on three different nodes, so two failures always leave one. Three may be too many: A, B and D together hold every copy of blocks 2 and 5. Three replicas protect against any two simultaneous failures, not against three.
The arithmetic, for a 200 MB file of the hospital's readings, readings.csv, with the slides' 64 MB blocks:
\[ n = \left\lceil \frac{S}{B} \right\rceil \]
\[ \text{disk} = r \cdot S \]
- \(S\)
- size of the file
- \(B\)
- block size: 64 MB on the slides, 128 MB since Hadoop 2
- \(r\)
- replication factor, 3 by default
- \(n\)
- number of blocks; the last one holds only what is left
200 / 64 = 3.1, so 4 blocks: three of 64 MB and a last one of 8 MB, which takes 8 MB of disk, not 64 — HDFS does not pad the last block (Apache documentation, not on the slides). With r = 3 the cluster stores 12 block replicas and 600 MB. The hook's 1 TB file: 16,384 blocks, 49,152 replicas, 3 TB.
"Goals of HDFS" (p. 850) puts numbers on the assumptions: a "Very Large Distributed File System" of "10K nodes, 100 million files, 10PB" that "Assumes Commodity Hardware" — "Files are replicated to handle hardware failure", "Detect failures and recover from them" — and is "Optimized for Batch Processing", with "Data locations exposed so that computations can move to where data resides".
Four decisions on one slide
"Distributed File System" (p. 851) names them:
- "Single Namespace for entire cluster": one directory tree, however many machines.
- "Data Coherency": "Write-once-read-many access model", "Client can only append to existing files". If nobody can overwrite a block, its three copies can never disagree, and the DataNodes need no protocol to keep them consistent.
- "Files are broken up into blocks": "Typically 64MB block size", "Each block replicated on multiple DataNodes".
- "Intelligent Client": "Client can find location of blocks", "Client accesses data directly from DataNode" — the subject of section 4.
Why such big blocks? HDFS "works best with a smaller number of large files" — "Millions as opposed to billions of files", "Typically 100MB or more per file" — and is "Optimized for streaming reads of large files and not random reads" (p. 847). The first reason is the disk; the argument below is a back-of-envelope one from Tom White's book, not from the slides.
A disk first seeks — it moves its head to the block — and then streams the bytes. Typical figures: seek time \(t_s \approx 10\ \text{ms}\), transfer rate \(v \approx 100\ \text{MB/s}\).
Reading one block of size \(B\) takes
\[ t = t_s + \frac{B}{v} , \]and only the second term moves data.
Ask the seek to cost at most 1 % of the transfer:
\[ t_s \le 0.01\,\frac{B}{v} \] \[ B \ge 100\, t_s\, v . \]With the typical figures:
\[ B \ge 100 \times 0.01\ \text{s} \times 100\ \text{MB/s} \] \[ B \ge 100\ \text{MB} . \]Check the defaults: a 64 MB block streams in 0.64 s, so the 10 ms seek is 1.5 % of the read; at 128 MB it is 0.8 %. A 4 KB block, the size an ordinary file system uses, streams in 0.04 ms: the disk would spend 99.6 % of its time seeking.
The NameNode, the one very important server
"The NameNode keeps track of which blocks make up a file and where they are stored" (p. 848). It "Manages File System Namespace": it "Maps a file name to a set of blocks" and "Maps a block to the DataNodes where it resides"; it also does "Cluster Configuration Management" and runs the "Replication Engine for Blocks" (p. 854). Two maps, then: readings.csv → blk 1, blk 2, blk 3, blk 4 and blk 1 → dn 1, dn 6, dn 5.
"NameNode Metadata" (p. 855): "The entire metadata is in main memory", with "No demand paging of metadata". It holds the "List of files", the "List of Blocks for each file", the "List of DataNodes for each block" and "File attributes, e.g. creation time, replication factor", plus "A Transaction Log" that "Records file creations, file deletions etc".
Memory is the second reason for few, large files. Every request starts with a lookup that must never wait for a disk, so the whole namespace lives in RAM. Tom White's rule of thumb (background) is about 150 bytes per file, directory or block: the 1 TB file in 64 MB blocks costs about 2.5 MB; the "100 million files" of p. 850, one block each, are 200 million objects, about 30 GB; the same terabyte as a billion 1 KB files would cost about 300 GB (computed).
The DataNode, a block server
A DataNode is "A Block Server" (p. 856): it "Stores data in the local file system (e.g. ext3)", "Stores metadata of a block (e.g. CRC)" and "Serves data and metadata to Clients". Its "Block Report" "Periodically sends a report of all existing blocks to the NameNode", and it "Facilitates Pipelining of Data": it "Forwards data to other specified DataNodes". It knows blocks, never files: only the NameNode knows that blk 2 belongs to readings.csv. The block reports are how the NameNode learns its second map, which it rebuilds from them at every start (Apache documentation).
The arrows of Fig. 3 are the division of labour. Between a client and the NameNode travel only "Metadata ops". Data travels between clients and DataNodes ("Read", "Write"). The NameNode sends "Block ops" to the DataNodes — copy this block, delete that one — and DataNodes copy blocks to each other ("Replication"), across racks. The deck's closing "HDFS Summary" (p. 888) says it in one line: "HDFS clients talk to the NameNode for metadata-related activities, and to DataNodes to read and write files".
The NameNode knows where every block is and stores none of them; the DataNodes store every block and know nothing about files.
Keeping every block alive
"Hardware will fail" is answered by four mechanisms: where the copies go, heartbeats with the replication engine, checksums, and the rebalancer. One failure they do not cover in HDFS 1: that of the NameNode itself.
Where the copies go
"Block Placement" (p. 857): "One replica on local node", "Second replica on a remote rack", "Third replica on same remote rack", "Additional replicas are randomly placed"; and "Clients read from nearest replicas". A rack is a cabinet of machines behind one switch and one power line, so it can fail as a whole.
If the three holders of a block failed independently, each with probability \(p\), losing the block would take all three: \(p^3\) — the independence of Bernoulli trials, and exactly what a rack failure breaks. Hence two racks. Why not three? Two already survive the loss of either rack, and the write pipeline of section 4 goes local node → remote rack → same remote rack, crossing the slow link between racks once per block; a third rack would cross it twice.
"The three replicas go to three different racks." No: the local node, then two nodes of one remote rack (p. 857).
Heartbeats and the replication engine
"DataNodes send hearbeat to the NameNode", "Once every 3 seconds", and the "NameNode uses heartbeats to detect DataNode failure" (p. 858): silence is the signal. The "Replication Engine" (p. 859) then acts: "NameNode detects DataNode failures", "Chooses new DataNodes for new replicas", "Balances disk usage", "Balances communication traffic to DataNodes". By default a DataNode is declared dead only after about ten minutes of silence (Apache defaults, not on the slides): a short network hiccup must not trigger terabytes of copying.
Checksums and the rebalancer
Failure also means silent corruption. "Data Correctness" (p. 860): "Use Checksums to validate data", "Use CRC32". At "File Creation" the "Client computes checksum per 512 bytes" and the "DataNode stores the checksum"; at "File access" the "Client retrieves the data and checksum from DataNode", and "If Validation fails, Client tries other replicas". That is the second job of the three copies. The price is a 4-byte CRC32 per 512 bytes: 4 / 512 ≈ 0.8 % more disk (computed).
The "Rebalancer" (p. 863; the Apache tool is called the balancer) has the "Goal: % disk full on DataNodes should be similar". It is "Usually run when new DataNodes are added", the "Cluster is online when Rebalancer is active", and the "Rebalancer is throttled to avoid network congestion". A new node arrives empty, and since map tasks go where the blocks are, a node without blocks gets no work until the rebalancer moves some onto it.
The failure that is not routine
A dead DataNode is routine. A dead NameNode is not: it is "A single point of failure in HDFS 1" (p. 861). HDFS 1 protects the namespace, not the service: the "Transaction Log stored in multiple directories", "A directory on the local file system" and "A directory on a remote file system (NFS/CIFS)". If the machine burns, the namespace can be loaded on another one — after an outage.
The transaction log only grows: every change is appended to it, while the FsImage — the saved snapshot of the namespace — is not rewritten as the NameNode runs. A long log means a slow restart. The "Secondary NameNode" (p. 864) keeps it short, as Fig. 4 shows: it "Copies FsImage and Transaction Log from Namenode to a temporary directory", "Merges FSImage and Transaction Log into a new FSImage in temporary directory", "Uploads new FSImage to the NameNode", and then the "Transaction Log on NameNode is purged". (The Apache documentation calls the log the EditLog.)
Before: the log keeps growing
After: a fresh FsImage, an empty log
"The Secondary NameNode takes over when the NameNode dies." No. It is a checkpointer: its copy of the namespace lags behind and it cannot serve clients. At best its last FsImage is a recent backup. A real standby arrives with Hadoop 2 (section 5).
Three copies on two racks, a heartbeat every 3 seconds and the replication engine make a dead DataNode a non-event; the NameNode is the one machine HDFS 1 cannot lose.
Reading and writing: the NameNode stays out of the way
"Data Retrieval" (p. 866) takes two moves: the client "Communicates with the NameNode to determine which blocks make up a file and on which data nodes those blocks are stored", "Then communicated directly with the data nodes to read the data". Writing uses a pipeline, "Data Pieplining" on p. 862: "Client retrieves a list of DataNodes on which to place replicas of a block", "Client writes block to the first DataNode", "The first DataNode forwards the data to the next node in the Pipeline", "When all replicas are written, the Client moves on to write the next block in file". The client uploads each block once, not three times.
The widget plays the two diagrams of the slides (pp. 867–868). Their boxes are Java classes of the HDFS client library: DistributedFileSystem talks to the NameNode, FSDataInputStream and FSDataOutputStream move the bytes in packets. The application itself only calls open or create, read or write, and close.
Why this design? The NameNode answers questions of a few bytes — where are the blocks, may this file be created — while every byte of data flows between clients and DataNodes, many pairs at once. That is the "Intelligent Client" of p. 851 and the "very high aggregate bandwidth" of p. 850.
Metadata goes to the NameNode, data goes to the DataNodes: a read asks where the blocks are and then streams them directly, a write is one upload into a pipeline.
Hadoop 2: YARN and the end of the lone masters
Hadoop 1 had a second very important server. MapReduce ran under one JobTracker for the whole cluster and a TaskTracker on every machine, which ran tasks in fixed map and reduce slots — the Master and workers of lesson 30, drawn on p. 832. The slides name them again on pp. 842 and 882 without defining them. The JobTracker did two jobs at once: it managed the resources of the cluster, and it scheduled and monitored every task of every job.
"Hadoop 1.0 Limitations" (p. 842) lists the cost. "Scalability": "Maximum Cluster Size – 4000 Nodes", "Maximum Concurrent Tasks – 40000", "Coarse synchronization in Job Tracker". "Single point of failure": "Failure kills all queued and running jobs", "Jobs need to be resubmitted by users". And "Restart is very tricky due to complex state". Fig. 5 is the slide's answer (p. 843).
Hadoop 1.0: one layer does two jobs
Hadoop 2.0: YARN takes resource management
YARN (p. 882) "Splits up the two major functions of JobTracker":
Resource Manager — the "Global Resource Manager", for "Cluster resource management"; one per cluster.
Application Master — "Job scheduling and monitoring (one per application)". It "negotiates resource containers from the Scheduler, tracking their status and monitoring for progress", and "itself runs as a normal container".
NodeManager — what replaces the TaskTracker: "A new per-node slave is responsible for launching the applications' containers, monitoring their resource usage (cpu, memory, disk, network) and reporting to the Resource Manager".
Container — a share of one machine's resources, some CPU and some memory, in which one process of an application runs: a map or a reduce task, a Spark executor, or the Application Master itself.
And "YARN maintains compatibility with existing MapReduce applications and users": MapReduce became one application among others. (The Apache documentation writes ResourceManager and ApplicationMaster as one word; this lesson keeps the slides' spelling.)
What the split buys (pp. 884–886):
- Scale. "Clusters of 6,000-10,000 machines", "100,000+ concurrent tasks", "10,000 concurrent jobs" (p. 884). Both slides allow about ten tasks per machine — 40000 / 4000 and 100,000 / 10,000 (computed) — so what grew is the number of machines one master can serve: the per-job work moved into the Application Masters.
- Fault tolerance (p. 885). For the Resource Manager, "No single point of failure – state saved in ZooKeeper" and "Application Masters are restarted automatically on RM restart"; for the Application Master, "Optional failover via application-specific checkpoint" and "MapReduce applications pick up where they left off via state saved in HDFS". This also answers "Restart is very tricky due to complex state": the Resource Manager's own state is small — which applications exist and what they hold — while each job's progress lives with its Application Master and in HDFS.
- Upgrades. "Wire Compatibility": "Old clients can talk to new servers", and "Rolling upgrades" update the cluster one machine at a time instead of stopping it.
- Other paradigms (p. 886). "Support for programming paradigms other than MapReduce (Multi tenancy)": Tez, HBase on YARN, Spark, Giraph and Storm, "Enabled by allowing the use of paradigm-specific application master" — "Run all on the same Hadoop cluster!"
The NameNode gets a standby
Hadoop 2 removes the single point of failure of HDFS too. Its architecture slide (p. 883) draws an "Active NameNode" with "All name space edits logged to shared NFS storage; single writer (fencing)", and a "Standby NameNode" that will "Read edit logs and applies to its own namespace". The DataNodes report to both, so the standby is always ready to take over; fencing makes sure only one NameNode ever writes the log. The same slide puts a DataNode and a NodeManager on every worker: containers run where the blocks are.
Core and ecosystem
The exam-guideline deck asks for "Hadoop architecture and components". Apache Hadoop itself is four modules: Hadoop Common (the shared libraries), HDFS, YARN and MapReduce. "Hadoop Components" (p. 838) lists more — "Apache Hadoop", "Apache Hive", "Apache Pig", "Apache HBase", "Apache Zookeeper", "Flume, Hue, Oozie, and Sqoop" — separate projects that run on the core and ship together. Vendors test and package the whole stack as distributions (p. 841): Cloudera's CDH, MapR, the Hortonworks Data Platform and the Oracle Big Data Appliance, each with a ready-made virtual machine.
Answer a "components" question with the core first — its masters and workers — and then the ecosystem by role.
YARN splits the JobTracker into one Resource Manager that hands out containers and one Application Master per application that runs its job; a NodeManager on every machine runs the containers.
Hands on HDFS: shell, security, defaults
The last part of the deck is a session on the Cloudera virtual machine (user cloudera, HDFS home /user/cloudera). No past paper asks for these commands, so this section shows them rather than drills them; every one of them makes a piece of the architecture visible. "There are two types of shell commands" (p. 873). User commands: hdfs dfs "runs filesystem commands on the HDFS" and hdfs fsck "runs a HDFS filesystem checking command". Administration commands: hdfs dfsadmin "runs HDFS administration commands". The general form is hdfs command [genericOptions] [commandOptions] (p. 879).
# p. 874: list the root directory of HDFS
hdfs dfs -ls /# p. 875: a directory tdata in your HDFS home
hdfs dfs -mkdir tdata# p. 875: upload a local file into it
hdfs dfs -copyFromLocal tutorials/data/geneva.csv tdata# p. 875: list your home recursively
hdfs dfs -ls -R# p. 875: go to the local folder that holds the original
cd tutorials/data/# p. 875: download the HDFS copy next to it, under another name
hdfs dfs -copyToLocal tdata/geneva.csv geneva.csv.hdfs# p. 875: two identical digests = byte for byte the same file
md5sum geneva.csv geneva.csv.hdfs# not on the slides: the directory that p. 876 writes into
hdfs dfs -mkdir tdataset# p. 876: write a file from standard input (the lone "-")
echo "blah blah blah" | hdfs dfs -put - tdataset/tfile.txt# p. 876: print it
hdfs dfs -cat tdataset/tfile.txt# p. 877: delete it
hdfs dfs -rm tdataset/tfile.txtThe upload, the download and md5sum make the round trip of the session: the file was cut into blocks, copied three times and put back together, and came out identical. On p. 876 the session writes into tdataset, a directory it never creates (it made tdata): -put does not create a missing parent directory, hence the extra -mkdir.
# p. 874: disk space used, human-readable (K, M, G)
hdfs dfs -du -h /# p. 874: one total (-s, summary) for a whole directory
hdfs dfs -du -s /hbase/data/hbase/namespace/# p. 876: the replication factor of the file (3)
hdfs dfs -stat "%r" tdata/geneva.csv# p. 876: owner, group and access control list
hdfs dfs -getfacl tdata/geneva.csv# p. 877: the blocks of the file and the DataNodes holding them
hdfs fsck /user/cloudera/tdata/geneva.csv -files -blocks -locations# p. 877: missing blocks and the files they belong to
hdfs fsck / -list-corruptfileblocksfsck prints the NameNode's two maps for one file: its blocks, and for each block the DataNodes of its replicas. It only reports; repairing is the replication engine's job.
# p. 878: capacity, used space and one entry per DataNode
hdfs dfsadmin -report# p. 878: the tree of racks and their nodes
hdfs dfsadmin -printTopology# p. 878: ask one DataNode for its information, like a ping
hdfs dfsadmin -getDatanodeInfo localhost:50020# p. 879: the NameNodes of the cluster
hdfs getconf -namenodes# p. 879: the folder where the NameNode keeps its FsImage on the Cloudera VM
cd /var/lib/hadoop-hdfs/cache/hdfs/dfs/name/current# p. 879: dump that FsImage to an XML file
hdfs oiv -i fsimage_0000000000000003388 -o /tmp/fsimage.xml -p XML-printTopology shows the racks that block placement relies on; oiv, the offline image viewer, turns the FsImage of Fig. 4 into readable XML.
The older slide of p. 865 writes the same tools as hadoop dfs and hadoop dfsadmin, still accepted but deprecated since Hadoop 2, with a web page at http://host:port/dfshealth.jsp (the Hadoop 1 address of the NameNode's status page).
"HDFS Security" (p. 869) has three layers. Who you are: "Simple – insecure way of using OS username to determine hadoop identity" or "Kerberos – authentication using kerberos ticket", "Set by hadoop.security.authentication=simple|kerberos". What you may do: "File and Directory permissions are same like in POSIX" — r, w and x for an owner, a group and everybody else — "enabled by default (dfs.permissions.enabled=true)". Exceptions: "ACLs are used for implemention permissions that differ from natural hierarchy of users and groups", "enabled by dfs.namenode.acls.enabled=true" — what -getfacl lists.
"HDFS Configuration" (p. 870) lives in /etc/hadoop/conf/hdfs-site.xml, with the "HDFS Defaults" "Block Size – 64 MB", "Replication Factor – 3" and "Web UI Port – 50070".
Other doors into HDFS (pp. 871, 880): the "Java API (DistributedFileSystem)", the very class of the read and write diagrams; the "C wrapper (libhdfs)"; the "HTTP protocol" (the NameNode's web page, http://quickstart.cloudera:50070); the "WebDAV protocol"; and the shell, "one of the simplest and most familiar". A FUSE mount makes HDFS look like a local directory, so that "standard Unix utilities such as 'ls', 'cd', 'cp', 'mkdir', 'find', 'grep'" work on it; 8020 is the port of the NameNode's RPC service:
# p. 880: a local directory to mount HDFS on
mkdir /home/cloudera/hdfs# p. 880: mount HDFS there
sudo hadoop-fuse-dfs dfs://quickstart.cloudera:8020 /home/cloudera/hdfsBack to the 3 TB file
Run fsck with -files -blocks -locations on the hook's file and HDFS answers with 16,384 blocks, each followed by three DataNode addresses. That is the 3 TB, and what it buys: any DataNode or any whole rack can fail, a corrupt copy is replaced by a good one, and the file never notices.
The NameNode keeps those 49,152 locations in RAM — a few megabytes — and answers every "where?" from memory. When a hundred analysts read the file, each asks the NameNode where the blocks are, a few bytes per block, and streams the blocks themselves from the DataNodes: a hundred network links at work and not one byte of data through the NameNode, which is why every machine you add adds bandwidth. The price of keeping all the knowledge in one machine is the single point of failure of HDFS 1; Hadoop 2's standby NameNode is the answer.
Exam-style questions
True/false, multiple choice and short open answers, the forms of the theory part, where a wrong answer may cost points; plus one small computation to check the block arithmetic. The exam-guideline deck lists "Hadoop architecture and components" among the theory questions, and "Hadoop simple examples" among the exercises — but none of the nine past papers has a Hadoop exercise.
True or false: in HDFS, a client that writes a file sends each block to the NameNode, which then stores it on three DataNodes.
Show model answer
False. The NameNode handles only metadata. The client calls create() and the NameNode adds the file to the namespace; for each block the client receives a list of DataNodes chosen by the placement policy. The client streams the block, in packets, to the first DataNode only; the first DataNode forwards the data to the second and the second to the third (data pipelining), and acknowledgements travel back along the pipeline. When all replicas are written the client moves on to the next block; at the end it closes the stream and the NameNode is told that the file is complete. Keeping the NameNode out of the data path is what gives HDFS its high aggregate bandwidth.
The trap is the NameNode placed next to the word "stores". A full answer adds the pipeline: one upload per block, three copies, acks flowing back.
Which of the following statements agree with the lecture's comparison between an RDBMS and Hadoop? Select all that apply.
Show model answer
The first, third and fifth statements.
A select-all question where a wrong tick may cost points. The two false options are the classic misreadings: the access column swapped, and "low integrity" read as "loses data".
A hospital heating management system stores its sensor readings in HDFS as one file of 200 MB. With the defaults printed on the slides — block size 64 MB, replication factor 3 — how many block replicas does the cluster store for this file?
Show model answer
\[ n = \left\lceil \frac{200}{64} \right\rceil = 4 \]
\[ 4 \times 3 = 12 \]
Four blocks (64 + 64 + 64 + 8 MB), three replicas each: 12 block replicas, which occupy 3 × 200 MB = 600 MB of disk.
Two traps: rounding 3.1 down to 3 blocks, and computing the disk as 4 × 64 × 3 = 768 MB — the last block holds only its 8 MB.
Describe the architecture of Hadoop and its main components, distinguishing the core of Hadoop from its ecosystem.
Show model answer
Hadoop is a software platform for storing and processing very large data sets in batch on clusters of commodity machines; it is an open-source implementation of Google's GFS and MapReduce. Its core has four parts: Hadoop Common (shared libraries), HDFS (storage), YARN (resource management, since Hadoop 2) and MapReduce (batch processing). HDFS has a master/worker architecture: one NameNode keeps the namespace in memory — the blocks of each file and the DataNodes of each block — while the DataNodes store the blocks (64 MB each on the slides, replicated three times), send heartbeats and block reports, and serve data directly to clients. YARN has one Resource Manager for the cluster's resources, a NodeManager on every machine that launches and monitors containers, and one Application Master per application that negotiates containers and schedules and monitors its tasks; in Hadoop 1 both roles belonged to the JobTracker, with a TaskTracker per node. Every worker runs a DataNode and a NodeManager, so computation can be moved to the data. Around the core, the ecosystem adds separate projects: Hive (SQL queries), Pig (scripting), HBase (a column store for random access), ZooKeeper (coordination), Sqoop (data exchange with relational databases), Flume (log collection), Oozie (workflows), Mahout (machine learning), Hue (web interface) and Ambari (cluster management), packaged by distributions such as Cloudera's CDH or the Hortonworks Data Platform.
The question uses the guideline's own words. Give the core first, with the role of each master and worker, then the ecosystem by role; listing names without roles is the weak answer.
What metadata does the HDFS NameNode keep, and where? What are the transaction log and the Secondary NameNode for? Why is the NameNode a single point of failure in HDFS 1, and how does Hadoop 2 address it?
Show model answer
The NameNode manages the file-system namespace: it maps each file name to its list of blocks and each block to the DataNodes holding its replicas, and keeps file attributes such as creation time and replication factor. The entire metadata is in main memory, with no demand paging, so every lookup is fast and the number of files is bounded by RAM. Every change to the namespace, such as a file creation or deletion, is recorded in a transaction log, while the FsImage is the saved snapshot of the namespace. The Secondary NameNode periodically copies the FsImage and the transaction log, merges them into a new FsImage and uploads it to the NameNode, whose log is then purged: it is a checkpointing helper that keeps the log short and the restart fast, not a standby. In HDFS 1 there is a single NameNode, so when it fails the whole file system is unavailable; storing the transaction log in several directories, one on a remote NFS/CIFS file system, only prevents losing the namespace. Hadoop 2 adds an Active and a Standby NameNode: the active one writes every edit to a shared edit log, the standby replays it, and the DataNodes report to both, so the standby can take over.
Four things are checked: the two maps in RAM, the log and the FsImage, the Secondary as a checkpointer — the most common wrong answer calls it a backup NameNode — and the standby of Hadoop 2.
What were the limitations of Hadoop 1.0, and how does YARN overcome them?
Show model answer
In Hadoop 1.0 a single JobTracker managed the resources of the whole cluster and also scheduled and monitored every task of every job, while a TaskTracker on each node ran the tasks in map and reduce slots. Scalability was limited to about 4000 nodes and 40000 concurrent tasks, because of coarse synchronization in the JobTracker; the JobTracker was a single point of failure whose failure killed all queued and running jobs, which users had to resubmit; restarting it was tricky because of its complex state; and only MapReduce could run. YARN splits the two functions of the JobTracker: a global Resource Manager does only cluster resource management, and a per-application Application Master, itself running in a container, negotiates resource containers from the scheduler and schedules and monitors its own job; the TaskTracker is replaced by a per-node NodeManager that launches containers, monitors their CPU, memory, disk and network use and reports to the Resource Manager. As a result clusters reach 6,000–10,000 machines and more than 100,000 concurrent tasks; the Resource Manager saves its state in ZooKeeper and restarts the Application Masters automatically, and MapReduce applications resume from state saved in HDFS; protocols are wire-compatible, allowing rolling upgrades; and other paradigms — Tez, Spark, Giraph, Storm, HBase — run on the same cluster through their own Application Masters, while existing MapReduce jobs keep working.
Pair each limitation of p. 842 with its answer: scalability with the split and the numbers of p. 884; the single point of failure and the tricky restart with the small Resource Manager state saved in ZooKeeper, the restarted Application Masters and the MapReduce state in HDFS; MapReduce-only with paradigm-specific Application Masters. Wire compatibility and rolling upgrades are the bonus point.
Check yourself
A file is written from a DataNode of rack 1 with r = 3 and the default placement. Then rack 2 loses power. What is the state of each block?
Why does HDFS put the third replica on the same remote rack as the second, rather than on a third rack?
A client reads a file of four blocks. Which description of the traffic is right?
Which of these follow from the write-once-read-many model of p. 851? Select all that apply.
Forbid the operation that would break consistency, and consistency is free: that is the trade HDFS makes.
You store ten million files of 1 KB each in HDFS, 10 GB in all. Compared with one 10 GB file, which resource runs out first?
What is the Secondary NameNode for?
On a YARN cluster the Application Master of a Spark job crashes while two MapReduce jobs are running. What happens?
Key take-aways
- Hadoop is HDFS (storage) plus MapReduce (batch processing) on many commodity machines, an open-source version of Google's GFS and MapReduce; its design follows six assumptions — hardware will fail, batch throughput over latency, huge files, high aggregate bandwidth, write-once-read-many, moving computation is cheaper than moving data.
- HDFS cuts every file into blocks (64 MB on the slides, 128 MB today) and keeps three replicas of each: \(\lceil S/B \rceil\) blocks and \(r \cdot S\) of disk, the last block unpadded.
- The NameNode holds the whole namespace in RAM — file → blocks, block → DataNodes — plus a transaction log; the DataNodes store the blocks, send a heartbeat every 3 seconds and block reports, and forward data in pipelines.
- Placement puts one replica on the local node and two on one remote rack; silent heartbeats start the replication engine; CRC32 per 512 bytes lets a client fall back to another replica; the rebalancer evens out the disks.
- A read asks the NameNode where the blocks are and then streams them from the DataNodes; a write goes through a pipeline of DataNodes with acks flowing back; no file data ever passes through the NameNode, which is why bandwidth grows with the cluster.
- The NameNode is the single point of failure of HDFS 1 and the Secondary NameNode only checkpoints; Hadoop 2 adds a standby NameNode, and YARN splits the JobTracker into a Resource Manager and one Application Master per application, and replaces the TaskTracker with a NodeManager per machine.
- Apache Hadoop's core is Common, HDFS, YARN and MapReduce; Hive, Pig, HBase, ZooKeeper, Flume, Sqoop, Oozie, Hue, Mahout and Ambari form the ecosystem around it.
What to watch and read
-
Distributed File Systems
Mining Massive Datasets, Stanford16 min
Why a cluster of cheap machines needs a file system of big replicated chunks and one master that knows where they are — the problem HDFS solves, told before any Hadoop name appears.
-
HDFS Architecture
Apache Hadoop
The design document the slides follow: the assumptions of p. 837, the original of the architecture figure of p. 853, and the placement rule of p. 857 in full. Its numbers are today's (128 MB blocks), not the slides'.
-
Apache Hadoop YARN
Apache Hadoop
One page and one figure: ResourceManager, NodeManager, ApplicationMaster and containers, the split of p. 882 in the project's own words. Read it next to the third widget.
-
The Hadoop Distributed File System, illustrated
Vu Trinh, 2024
A walk through the original HDFS paper with a diagram for each mechanism: block reports, the write pipeline, replica placement, checkpointing, the balancer — the slides' bullet points as pictures.
-
Lecture 3: GFS
MIT 6.824 Distributed Systems
The Google file system HDFS copied, discussed with the exam's "why" questions: why one master, why big chunks, what the master keeps in memory, what happens when a server dies.