HDFS Tutorial: Architecture, Read & Write Operation

What is HDFS?

HDFS is a distributed file system for storing very large data files, running on clusters of commodity hardware. It is fault tolerant, scalable, and extremely simple to expand. Hadoop comes bundled with HDFS (Hadoop Distributed File System).

When data exceeds the capacity of storage on a single physical machine, it becomes essential to divide it across a number of separate machines. A file system that manages storage specific operations across a network of machines is called a distributed file system. HDFS is one such software.

HDFS Architecture

An HDFS cluster primarily consists of a NameNode that manages the file system metadata and DataNodes that store the actual data.

• NameNode: The NameNode can be considered as the master of the system. It maintains the file system tree and the metadata for all the files and directories present in the system. Two files, the ‘namespace image’ and the ‘edit log’, are used to store metadata information. The NameNode has knowledge of all the DataNodes containing data blocks for a given file, however, it does not store block locations persistently. This information is reconstructed every time from the DataNodes when the system starts.
• DataNode: DataNodes are slaves which reside on each machine in a cluster and provide the actual storage. They are responsible for serving read and write requests from the clients.

Because a lone NameNode would be a single point of failure, current clusters run an active NameNode alongside a standby NameNode that shares the edit log through a quorum of JournalNodes, with automatic failover between the two.

Read and write operations in HDFS operate at a block level. Data files in HDFS are broken into block-sized chunks, which are stored as independent units. The default block size is 64 MB in Hadoop 1.x. From Hadoop 2.x onward the default dfs.blocksize is 128 MB.

HDFS operates on a concept of data replication wherein multiple replicas of data blocks are created and are distributed on nodes throughout a cluster to enable high availability of data in the event of node failure. The default replication factor is three (dfs.replication), and when a DataNode stops sending heartbeats the NameNode automatically re-replicates its blocks elsewhere.

Do you know? A file in HDFS, which is smaller than a single block, does not occupy a block’s full storage.

Read Operation In HDFS

A data read request is served by HDFS, the NameNode, and the DataNodes. Let us call the reader a ‘client’. The diagram below depicts the file read operation in Hadoop.

1. A client initiates a read request by calling the ‘open()’ method of the FileSystem object; it is an object of type DistributedFileSystem.
2. This object connects to the NameNode using RPC and gets metadata information such as the locations of the blocks of the file. Please note that these addresses are of the first few blocks of a file.
3. In response to this metadata request, the addresses of the DataNodes having a copy of that block are returned.
4. Once addresses of DataNodes are received, an object of type FSDataInputStream is returned to the client. FSDataInputStream contains DFSInputStream which takes care of interactions with the DataNode and the NameNode. In step 4 shown in the above diagram, a client invokes the ‘read()’ method which causes DFSInputStream to establish a connection with the first DataNode holding the first block of a file.
5. Data is read in the form of streams wherein the client invokes the ‘read()’ method repeatedly. This process of the read() operation continues till it reaches the end of the block.
6. Once the end of a block is reached, DFSInputStream closes the connection and moves on to locate the next DataNode for the next block.
7. Once the client has finished reading, it calls the close() method.

Write Operation In HDFS

In this section, we will understand how data is written into HDFS through files. The diagram below traces that write path.

1. A client initiates a write operation by calling the ‘create()’ method of the DistributedFileSystem object which creates a new file – Step no. 1 in the above diagram.
2. The DistributedFileSystem object connects to the NameNode using an RPC call and initiates new file creation. However, this file creation operation does not associate any blocks with the file. It is the responsibility of the NameNode to verify that the file (which is being created) does not exist already and that the client has correct permissions to create a new file. If a file already exists or the client does not have sufficient permission to create a new file, then an IOException is thrown to the client. Otherwise, the operation succeeds and a new record for the file is created by the NameNode.
3. Once a new record in the NameNode is created, an object of type FSDataOutputStream is returned to the client. A client uses it to write data into HDFS. The data write method is invoked (step 3 in the diagram).
4. FSDataOutputStream contains a DFSOutputStream object which looks after communication with the DataNodes and the NameNode. While the client continues writing data, DFSOutputStream continues creating packets with this data. These packets are enqueued into a queue which is called the DataQueue.
5. There is one more component called DataStreamer which consumes this DataQueue. DataStreamer also asks the NameNode for allocation of new blocks, thereby picking desirable DataNodes to be used for replication.
6. Now, the process of replication starts by creating a pipeline using DataNodes. In our case, we have chosen a replication level of 3 and hence there are 3 DataNodes in the pipeline.
7. The DataStreamer pours packets into the first DataNode in the pipeline.
8. Every DataNode in a pipeline stores the packet received by it and forwards the same to the second DataNode in the pipeline.
9. Another queue, the ‘Ack Queue’, is maintained by DFSOutputStream to store packets which are waiting for acknowledgment from the DataNodes.
10. Once acknowledgment for a packet in the queue is received from all DataNodes in the pipeline, it is removed from the ‘Ack Queue’. In the event of any DataNode failure, packets from this queue are used to reinitiate the operation.
11. After a client is done writing data, it calls the close() method (Step 9 in the diagram). The call to close() results in flushing the remaining data packets to the pipeline, followed by waiting for acknowledgment.
12. Once a final acknowledgment is received, the NameNode is contacted to tell it that the file write operation is complete.

Access HDFS Using the Java API

In this section, we try to understand the Java interface used for accessing Hadoop’s file system.

In order to interact with Hadoop’s filesystem programmatically, Hadoop provides multiple Java classes. The package named org.apache.hadoop.fs contains classes useful in manipulation of a file in Hadoop’s filesystem. These operations include open, read, write, and close. The Hadoop file API is generic and can be extended to interact with filesystems other than HDFS.

Reading a file from HDFS, programmatically

Object java.net.URL is used for reading contents of a file. To begin with, we need to make Java recognize Hadoop’s hdfs URL scheme. This is done by calling the setURLStreamHandlerFactory method on the URL object and passing an instance of FsUrlStreamHandlerFactory to it. This method needs to be executed only once per JVM, hence it is enclosed in a static block.

An example code is-

public class URLCat {
    static {
        URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
    }
    public static void main(String[] args) throws Exception {
        InputStream in = null;
        try {
            in = new URL(args[0]).openStream();
            IOUtils.copyBytes(in, System.out, 4096, false);
        } finally {
            IOUtils.closeStream(in);
        }
    }
}

This code opens and reads the contents of a file. The path of this file on HDFS is passed to the program as a command line argument.

Access HDFS Using the Command-Line Interface

This is one of the simplest ways to interact with HDFS. The command-line interface has support for filesystem operations like reading a file, creating directories, moving files, deleting data, and listing directories.

We can run ‘$HADOOP_HOME/bin/hdfs dfs -help’ to get detailed help on every command. Here, ‘dfs’ is a shell command of HDFS which supports multiple subcommands. In current Hadoop releases hdfs dfs is the preferred form, while the older hadoop fs command performs the same work for any supported file system.

Some of the widely used commands are listed below along with some details of each one.

1. Copy a file from the local filesystem to HDFS

$HADOOP_HOME/bin/hdfs dfs -copyFromLocal temp.txt /

This command copies the file temp.txt from the local filesystem to HDFS, as the output below shows.

2. We can list files present in a directory using -ls

$HADOOP_HOME/bin/hdfs dfs -ls /

In the listing below we can see the file ‘temp.txt’ (copied earlier) under the ‘ / ‘ directory.

3. Command to copy a file to the local filesystem from HDFS

$HADOOP_HOME/bin/hdfs dfs -copyToLocal /temp.txt

This command mirrors -get and accepts an explicit local destination path as a second argument. The output below shows temp.txt copied to the local filesystem.

4. Command to create a new directory

$HADOOP_HOME/bin/hdfs dfs -mkdir /mydirectory

The command completes silently, as the prompt below shows.

Check whether the directory is created or not. Now, you should know how to do it 😉

FAQs