CCS334 BIG DATA ANALYTICS

UNIT III

HADOOP BASIC TOOLS

Hadoop Data Format

A storage format is just a way to define how information is stored in a file. When
dealing with Hadoop’s file system not only do you have all of these traditional
storage formats available to you (like you can store PNG and JPG images on HDFS
if you like), but you also have some Hadoop-focused file formats to use for
structured and unstructured data. A huge issue with HDFS-enabled applications like
MapReduce and Spark is the time it takes to find relevant data in a particular
location and the time it takes to write the data back to another location. Choosing an
appropriate file format can have some significant benefits.

Benefits from Selecting Appropriate Format

1. Faster reads
2. Faster writes
3. Splittable files ( when we need to read a part of file and not the complete )
4. Schema changes or evolution
5. Compression support

Different Data Formats

1. Avro
2. Parquet
3. ORC
4. Arrow (Incubating)

1
 CCS334 BIG DATA ANALYTICS

Text file Format

Simple text-based files are common in the non-Hadoop world, and they’re super
common in the Hadoop world too. Data is laid out in lines, with each line being a
record. Lines are terminated by a newline character “\n”

Text-files are inherently splittable (just split on \n characters!), but if you want to
compress them you’ll have to use a file-level compression codec that support
splitting, such as BZIP2/

Because these files are just text files you can encode anything you like in a line of
the file. One common example is to make each line a JSON document to add some
structure. While this can waste space with needless column headers, it is a simple
way to start using structured data in HDFS. To Summarise,

1. Simple format
2. Data is laid out in lines
3. Lines are terminated by \n
4. Format is inherently splittable

Sequence File Format

Sequence files were originally designed for MapReduce, so the integration is

smooth. They encode a key and a value for each record. Records are stored in a
binary format that is smaller than a text-based format.

Sequence files by default use Hadoop’s writable interface in order to figure out how
to serialize and deserialize classes to the file.

2
 CCS334 BIG DATA ANALYTICS

One benefit of sequence files is that they support block-level compression, so you
can compress the contents of the file while also maintaining the ability to split the
file into segments for multiple map tasks.

Sequence files are well supported across Hadoop and many other HDFS enabled
projects, To Summarise,

1. They encode a key and value for each record

2. Records are stored in binary format.
3. Support block level compression
4. Well supported across hadoop eco-system

Avro

Avro is an opinionated format which understands that data stored in HDFS is

usually not a simple key/value combo like int or string. The format encodes the
schema of its contents directly in the file which allows you to store complex objects
natively. Avro not really feels like a file format, it’s a file format plus a serialization
and deserialization framework.

Avro is a well thought out format which defines file data schemas in JSON (for
interoperability), allows for multiple serialization/deserialization. It also supports
block-level compression. For most Hadoop-based use cases Avro is a really good
choice. To Summarise,

1. Language-neutral data serialization system

2. Schema is always present
3. Schema is in JSON
4. Data is binary encoded

3
 CCS334 BIG DATA ANALYTICS

5. Datafile has a metadata section where schema is stored

6. Supports compression
7. Are splittable
8. Supported by Pig, Hive, Crunch, Spark

Parquet

The latest buzz in file formats for Hadoop is columnar file storage. That means that
instead of just storing rows of data adjacent to one another you also store column
values adjacent to each other. So datasets are partitioned both horizontally and
vertically.

One huge benefit of columnar oriented file formats is that data in the same column
tends to be compressed together which can yield some massive storage
optimizations (as data in the same column tends to be similar).

Overall these formats can drastically optimize workloads, especially for Hive and
Spark which tend to just read segments of records rather than the whole thing
(which is more common in MapReduce). To Summarise,

1. Columnar format
2. Homogeneous data comes together, better for compression
3. Not good if queries often need the entire row

ORC

ORC stands for Optimized Row Columnar which means it can store data in an
optimized way than the other file formats. ORC reduces the size of the original data
up to 75%. As a result the speed of data processing also increases and shows better
performance than Text, Sequence and RC file formats.
4
 CCS334 BIG DATA ANALYTICS

An ORC file contains rows data in groups called as Stripes along with a file footer.
ORC format improves the performance when Hive is processing the data. However,
the ORC file increases CPU overhead by increasing the time it takes to decompress
the relational data. ORC File format feature comes with the Hive 0.11 version and
cannot be used with previous versions.

With this, we come to end of this article, I hope, this serves as quick reference guide
for revisiting the popular Hadoop Data Storage Formats.

Hadoop Storage Format

A massive bottleneck for an HDFS-enabled application like Spark and MapReduce

is time taken to find relevant data in a particular location and to write the data back
to another location. Issues exacerbated with the difficulties managing large
datasets, such as evolving storage constraints or schemas. Here is various file
formats have involved as a way to ease these issues across some use cases.

Here are some significant benefits of choosing an appropriate file format -

 Faster read times

 Faster write times
 Splittable files
 Schema evolution support
 Advanced compression support

Some file formats designed for general use (like Spark or MapReduce), others are
designed for more specific use cases, and some designed with specific data
characteristics in mind. So there is a lot of choices.

How to Adopt File Format?

5
 CCS334 BIG DATA ANALYTICS

There three types of performance to consider when choosing a file format.

 Write performance - how to write data fast.

 Partial read performance - how to read individual column within a file.
 Full read performance - How to read every data element in a file.

A columnar, compressed file format like ORC or Parquet used for partial and full
read performance, but these are expensive of write performance. Uncompressed
CSV files fast to write but lack column-orientation and slow compression slow for
reads.

Many options can be used to store data. If need to storing intermediate data
between MapReduce jobs, then Sequence files are preferred. If query performance
is most important, then ORC or Parquet are optimal, but these files take longer to
write. If the schema is going to change over time, then Avro is best, but query
performance will be slower than ORC or Parquet. CSV files are best if extract data
from Hadoop to bulk load into a database.

Benefits of Hadoop Storage Format

The benefits of Hadoop Storage Format are listed below:

 Faster read times

 Faster write times
 Split files
 Schema evolution support (allowing you to change the fields in a data set).
 Advanced compression support (compress the files with a compression codec
without sacrificing these features).

Best practices of Hadoop File Storage

6
 CCS334 BIG DATA ANALYTICS

 When the need to accessing an only a small subset of columns then used a
columnar data format.
 When necessary to obtaining many columns then used a row-oriented database
instead of a columnar database.
 If schema changes over time then use Avro instead of ORC or Parquet.
 If need to perform query then use ORC or Parquet instead of Avro.
 If need to perform column add operation then use Parquet instead of ORC.

Scaling Up (Vertical Scaling)

Scaling up (or vertical scaling) is adding more resources—like CPU, memory, and
disk—to increase more compute power and storage capacity. This term applies to
traditional applications deployed on physical servers or virtual machines as well.

The diagram above shows an application pod that begins with a small
configuration with 1 CPU, 2 GB of memory, and 100 GB disk space and scales

7
 CCS334 BIG DATA ANALYTICS

vertically to large configurations with 4 CPU, 8 GB of memory, and 500 GB disk

space. Now with more compute resources and storage space, this application can
process and serve more requests from clients.

Scaling up seems to be a good choice if your application only needs to scale to a

reasonable size. There are some advantages and disadvantages of scaling up:

Advantages

 It is simple and straightforward. For the applications with more traditional and
monolithic architecture, it is much simpler to just add more compute resources to
scale.
 You can take advantage of powerful server hardware. Today’s servers are more
powerful than ever, with more efficient CPUs, larger DIMM capacities, faster
disks, and high-speed networking. By taking advantage of these ample compute
resources, you can scale up to very large application pods.

Disadvantages

Scaling up has limits.

Even with today’s powerful servers, as you continue to add compute resources to
your application pod, you will still hit the physical hardware limitations sooner or
later.

 Bottlenecks develop in compute resources. As you add compute resources to a

physical server, it is difficult to increase and balance the performance linearly for
all the components, and you will most likely hit a bottleneck somewhere. For
example, initially your server has a memory bottleneck with 100% usage of

8
 CCS334 BIG DATA ANALYTICS

memory and 70% usage of CPU. After doubling the number of DIMMs, now you
have 100% of CPU usage vs 80% of memory usage.
 It may cost more to host applications. Usually the larger servers with high
compute power cost more. If your application requires high compute resources,
using these high-cost larger servers may be the only choice.

With physical hardware limitations, scaling up vertically is a rather short term

solution if your application needs to continue growing.

Scaling Out (Horizontal Scaling)

Scaling out (or horizontal scaling) addresses some of the limitations of the scale up
method. With horizontal scaling, the compute resource limitations from physical
hardware are no longer the issue. In fact, you can use any reasonable size of server
as long as the server has enough resources to run the pods. The diagram below
shows an example of an application pod with three replicas scaling out to five
replicas, and this is how Kubernetes normally manages application workloads.

Scaling out also has its advantages and disadvantages:

Advantages

9
 CCS334 BIG DATA ANALYTICS

 It delivers long-term scalability. The incremental nature of scaling out allows you
to scale your application for expected and long-term growth.
 Scaling back is easy. Your application can easily scale back by reducing the
number of pods when the load is low. This frees up compute resources for other
applications.
 You can utilize commodity servers. Normally, you don’t need large servers to run
containerized applications. Since application pods scale horizontally, servers can
be added as needed.

Disadvantages

 It may require re-architecting. You will need to re-architect your application if

your application is using monolithic architecture(s).

Scaling Stateless Applications

Stateless applications do not store data in the application, so they can be used as
short-term workers. Kubernetes manages stateless applications very well. In
Kubernetes, a HorizontalPodAutoscaler automatically updates a workload
resource, such as a Deployment, with the aim of automatically scaling the
workload to match demand. This means if the load of application pods increases,
the HorizontalPodAutoscaler keeps increasing the number of pods until the load
comes back to normal range.

If the load decreases and the number of pods is above the configured minimum, the
Horizontal Pod Autoscaler instructs the Deployment to scale back down.

The Horizontal Pod Autoscaler is implemented as a Kubernetes API resource and a

controller. The resource determines the behavior of the controller. The horizontal

10
 CCS334 BIG DATA ANALYTICS

pod autoscaling controller, running within the Kubernetes control plane,

periodically adjusts the desired scale of its target Deployment to match observed
metrics such as average CPU utilization, average memory utilization, or any other
custom metric you specify.

Kubernetes implements horizontal pod autoscaling as a control loop that runs

intermittently. The interval is set by the –horizontal-pod-autoscaler-sync-period
parameter to the kube-controller-manager, and the default interval is 15 seconds.

Scaling Stateful Applications

Unlike stateless applications, stateful applications such as databases need to store

data in persistent volumes. This makes it more difficult for Kubernetes to manage
stateful applications. The storage layer itself is a very complex topic, and we will
not discuss it in detail in this post.

For some stateful applications, you may be able to configure Kubernetes’

HorizontalPodAutoscaler to automate scale out operations, but once data is written,
it is difficult to scale back down.

For example, if a database gets larger and storage capacity usage gets high, we
need to scale up or expand the persistent volume. This operation is usually
performed separately from scaling up the compute resource by adding CPU and
memory. In this section, we will focus on scale up and scale out of stateful
database pods.

Scale Up Database Pods Vertically

When you use relational databases like MySQL or PostgreSQL, you can create
multiple nodes of database pods to form a cluster to increase high availability and
11
 CCS334 BIG DATA ANALYTICS

scale the performance. For example, when you create a 3-node cluster of MySQL,
one of the nodes is the primary node that accepts read and write requests. Two
other nodes are called read-replicas, and they only serve the read requests. Since
you can only write to the primary node, data consistency can be maintained in the
cluster.

If the demand of read requests increases and write requests stay the same, then
you can scale out your database by adding more read replica pods to the cluster.
However, if the write requests increase, adding more read replica pods will not
help since you can only write to the primary node. In this case, it is much simpler
just to scale up your database pod by adding more compute resources. There is a
way—called “sharding”—to split a database instance into multiple instances when
the database gets too large, but this introduces another level of complexity into the
cluster architecture.

12
 CCS334 BIG DATA ANALYTICS

Scale Out Database Pods Horizontally

If you are using a NoSQL distributed database like Cassandra, you can easily scale
out horizontally to meet your demand. Every Cassandra node can perform read and
write operations, and this makes a Cassandra cluster a masterless, or peer to peer,
architecture. In other words, distributed databases are made to scale very large.

HADOOP STREAMING

Introduction to Hadoop Streaming

Hadoop Streaming uses UNIX standard streams as the interface

between Hadoop and your program so you can write MapReduce program in any
language which can write to standard output and read standard input. Hadoop
offers a lot of methods to help non-Java development.

The primary mechanisms are Hadoop Pipes which gives a native C++ interface to
Hadoop and Hadoop Streaming which permits any program that uses standard
input and output to be used for map tasks and reduce tasks.

With this utility, one can create and run MapReduce jobs with any executable or
script as the mapper and/or the reducer.

Features of Hadoop Streaming

Some of the key features associated with Hadoop Streaming are as follows :

 Hadoop Streaming is a part of the Hadoop Distribution System.

 It facilitates ease of writing Map Reduce programs and codes.

13
 CCS334 BIG DATA ANALYTICS

 Hadoop Streaming supports almost all types of programming languages such as

Python, C++, Ruby, Perl etc.
 The entire Hadoop Streaming framework runs on Java. However, the codes might
be written in different languages as mentioned in the above point.
 The Hadoop Streaming process uses Unix Streams that act as an interface between
Hadoop and Map Reduce programs.
 Hadoop Streaming uses various Streaming Command Options and the two
mandatory ones are – -input directoryname or filename and -output directoryname

Hadoop Streaming architecture

As it can be clearly seen in the diagram above that there are almost 8 key parts in a
Hadoop Streaming Architecture. They are :

 Input Reader/Format
 Key Value
 Mapper Stream
 Key-Value Pairs
 Reduce Stream
 Output Format

14
 CCS334 BIG DATA ANALYTICS

 Map External
 Reduce External

The involvement of these components will be discussed in detail when we explain

the working of the Hadoop streaming. However, to precisely summarize the
Hadoop Streaming Architecture, the starting point of the entire process is when the
Mapper reads the input value from the Input Reader Format. Once the input data is
read, it is mapped by the Mapper as per the logic given in the code. It then passes
through the Reducer stream and the data is transferred to the output after data
aggregation is done. A more detailed description is given in the below section on
the working of the Hadoop Streaming.

How does Hadoop Streaming Work?

 Input is read from standard input and the output is emitted to standard output by
Mapper and the Reducer. The utility creates a Map/Reduce job, submits the job to
an appropriate cluster, and monitors the progress of the job until completion.
 Every mapper task will launch the script as a separate process when the mapper is
initialized after a script is specified for mappers. Mapper task inputs are converted
into lines and fed to the standard input and Line oriented outputs are collected from
the standard output of the procedure Mapper and every line is changed into a key,
value pair which is collected as the outcome of the mapper.
 Each reducer task will launch the script as a separate process and then the reducer
is initialized after a script is specified for reducers. As the reducer task runs,
reducer task input key/value pairs are converted into lines and fed to the standard
input (STDIN) of the process.

15
 CCS334 BIG DATA ANALYTICS

 Each line of the line-oriented outputs is converted into a key/value pair after it is
collected from the standard output (STDOUT) of the process, which is then
collected as the output of the reducer.
 Some Hadoop Streaming Commands

Option Description

-input directory_name or
Input location for the mapper.
filename

-output directory_name Input location for the reducer.

-mapper executable or
The command to be run as the mapper
JavaClassName

-reducer executable or
The command to be run as the reducer
script or JavaClassName

Make the mapper, reducer, or combiner

-file file-name executable available locally on the compute
nodes

By default, TextInputformat is used to return the

-inputformat
key-value pair of Text class. We can specify our
JavaClassName
class but that should also return a key-value pair.

16
 CCS334 BIG DATA ANALYTICS

Option Description

By default, TextOutputformat is used to take key-

-outputformat
value pairs of Text class. We can specify our
JavaClassName
class but that should also take a key-value pair.

-partitioner JavaClassName
The Class that determines which key to reduce.

-combiner
streamingCommand or The Combiner executable for map output
JavaClassName

-verbose The Verbose output.

-numReduceTasks It Specifies the number of reducers.

-mapdebug Script to call when map task fails

-reducedebug Script to call when reduce task fails

HADOOP PIPELINE

DATA PIPELINE IN HADOOP

A data pipeline is an arrangement of elements connected in series that is designed

to process the data in an efficient way. In this arrangement, the output of one
element is the input to the next element.
17
 CCS334 BIG DATA ANALYTICS

There are different components in the Hadoop ecosystem for different purposes.

Consider an application where you have to get input data from a CSV file, store it
hdfs, process it, and then provide the output. Here, you will first have to import
data from CSV file to hdfs using hdfs commands. Then you might have to use
MapReduce to process the data. To store data, you can use SQL or NoSQL
database such as HBase. To query the data you can use Pig or Hive. And if you
want to send the data to a machine learning algorithm, you can use Mahout.

These are some of the tools that you can use to design a solution for a big data
problem statement. When you integrate these tools with each other in series and
create one end-to-end solution, that becomes your data pipeline!

Now that you know what a data pipeline is, let me tell you about the most common
types of big data pipelines.

Types of Big Data Pipelines

When you create a data pipeline, it’s mostly unique to your problem statement. But
here are the most common types of data pipeline:

 Batch processing pipeline

 Real-time data pipeline
 Cloud-native data pipeline

Let’s discuss each of these in detail.

18
 CCS334 BIG DATA ANALYTICS

Batch Processing Pipeline

In this type of pipeline, you will be sending the data into the pipeline and process it
in parts, or batches. This type of pipeline is useful when you have to process a
large volume of data, but it is not necessary to do so in real time.

For example, suppose you have to create a data pipeline that includes the study and
analysis of medical records of patients. If you are using patient data from the past
20 years, that data becomes huge. But it is not necessary to process the data in real
time because the input data was generated a long time ago.

Real-Time Data Pipeline

You will be using this type of data pipeline when you deal with data that is being
generated in real time and the processing also needs to happen in real time.

For example, stock market predictions. There are different tools that people use to
make stock market predictions. To design a data pipeline for this, you would have
to collect the stock details in real-time and then process the data to get the output.

Cloud-Native Data Pipeline

Cloud helps you save a lot of money on resources. It prevents the need to have
your own hardware. In the cloud-native data pipeline, the tools required for the
data pipeline are hosted on the cloud. This is useful when you are using data stored
in the cloud. You can easily send the data that is stored in the cloud to the pipeline,
which is also on the cloud.

19
 CCS334 BIG DATA ANALYTICS

Components of a Hadoop Data Pipeline

a data pipeline is a combination of tools. These tools can be placed into different
components of the pipeline based on their functions. The three main components of
a data pipeline are:

 Storage component
 Compute component
 Message component

Storage Component

Because you will be dealing with data, it’s understood that you’ll have to use a
storage component to store the data. This storage component can be used to store
the data that is to be sent to the data pipeline or the output data from the pipeline.

When it comes to big data, the data can be raw. You can’t expect the data to be
structured, especially when it comes to real-time data pipelines. To handle
situations where there’s a stream of raw, unstructured data, you will have to use
NoSQL databases. The most important reason for using a NoSQL database is that
it is scalable. If you have used a SQL database or are using a SQL database, you
will see that the performance decreases when the data increases. NoSQL works in
such a way that it solves the performance issue.

Some of the most-used storage components for a Hadoop data pipeline are:

 Apache Cassandra
 Apache HBase
 HDFS

20
 CCS334 BIG DATA ANALYTICS

Compute Component

This component is where data processing happens. You are using the data pipeline
to solve a problem statement. And for that, you will be using an algorithm. The
execution of that algorithm on the data and processing of the desired output is
taken care by the compute component.

In Hadoop pipelines, the compute component also takes care of resource allocation
across the distributed system. You can consider the compute component as the
brain of your data pipeline.

Some of the most-used compute component tools are:

 MapReduce
 Apache Spark
 Apache Storm

Message Component

The message component plays a very important role when it comes to real-time
data pipelines. Messaging means transferring real-time data to the pipeline. Some
of the most used message component tools are:

 Apache Kafka
 Apache Pulsar

Hadoop is a distributed computing framework that allows for the processing of

large data sets across a cluster of computers. Its MapReduce paradigm allows for
the parallel processing of data, which is essential for dealing with big data. Hadoop
Pipes is an API that allows developers to write MapReduce programs in C++.

21
 CCS334 BIG DATA ANALYTICS

However, when working with large data records, the default memory limit of
Hadoop Pipes can cause problems. This blog post will explain how to pass large
data records to Map/Reduce tasks using Hadoop Pipes.

HADOOP PIPES

Hadoop Pipes is a C++ API that enables developers to write MapReduce programs
for Hadoop. The API allows the use of existing C++ libraries with Hadoop and
enables the development of high-performance MapReduce programs. Hadoop
Pipes is a popular choice for developers who need to process large data sets
efficiently.

The Problem with Large Data Records

When working with large data records, developers may encounter memory-related
issues. The default memory limit of Hadoop Pipes is 200 MB, which can be
insufficient for processing large data sets. If the data record size exceeds the
memory limit, the MapReduce program may fail.

How to Pass Large Data Records to Map/Reduce Tasks

To pass large data records to Map/Reduce tasks using Hadoop Pipes, developers
need to modify the Hadoop Pipes program to use external memory. External
memory is memory that is not managed by Hadoop and is instead managed by the
operating system. By using external memory, developers can process large data
sets without encountering memory-related issues.

To use external memory with Hadoop Pipes, developers need to use a technique
called memory mapping. Memory mapping is a technique that allows developers to
map external memory into the virtual memory of the program. This technique
22
 CCS334 BIG DATA ANALYTICS

enables the program to access the external memory as if it were part of the
program’s memory. Memory mapping is an efficient way to access external
memory because it eliminates the need to copy data between the program’s
memory and the external memory.

To use memory mapping with Hadoop Pipes, developers need to do the following:

1. Open the input file using the open function and obtain the file descriptor.
2. Map the input file into memory using the mmap function. The mmap function
returns a pointer to the mapped memory.
3. Read the data from the mapped memory.
4. Process the data.
5. Unmap the memory using the munmap function.
6. Close the input file using the close function.

Here is an example of how to use memory mapping with Hadoop Pipes:

#include <sys/mman.h>

#include <fcntl.h>

#include <unistd.h>

int main(int argc, char** argv) {

// Open the input file

int fd = open(argv[1], O_RDONLY);

23
 CCS334 BIG DATA ANALYTICS

// Map the input file into memory

char* data = (char*) mmap(NULL, 1024 * 1024 * 1024, PROT_READ,

MAP_PRIVATE, fd, 0);

// Process the data

process_data(data);

// Unmap the memory

munmap(data, 1024 * 1024 * 1024);

// Close the input file

close(fd);

return 0;

In this example, the input file is opened using the open function, and the file
descriptor is obtained. The input file is then mapped into memory using
the mmap function. The mmap function maps the entire file into memory.
The process_data function is called to process the data. The memory is then
unmapped using the munmap function, and the input file is closed using
the close function.

24
 CCS334 BIG DATA ANALYTICS

HDFS- Hadoop Distributed File system

Hadoop Distributed File system – HDFS is the world’s most reliable storage
system. HDFS is a Filesystem of Hadoop designed for storing very large files
running on a cluster of commodity hardware. It is designed on the principle of
storage of less number of large files rather than the huge number of small files.

Hadoop HDFS provides a fault-tolerant storage layer for Hadoop and its other
components. HDFS Replication of data helps us to attain this feature. It stores data
reliably, even in the case of hardware failure. It provides high throughput access to
application data by providing the data access in parallel.

HDFS Architecture:

Apache HDFS or Hadoop Distributed File System is a block-structured file

system where each file is divided into blocks of a pre-determined size. These
blocks are stored across a cluster of one or several machines. Apache Hadoop
HDFS Architecture follows a Master/Slave Architecture, where a cluster comprises

25
 CCS334 BIG DATA ANALYTICS

of a single NameNode (Master node) and all the other nodes are DataNodes (Slave
nodes). HDFS can be deployed on a broad spectrum of machines that support Java.
Though one can run several DataNodes on a single machine, but in the practical
world, these DataNodes are spread across various machines.

NameNode:

NameNode is the master node in the Apache Hadoop HDFS Architecture that
maintains and manages the blocks present on the DataNodes (slave
nodes). NameNode is a very highly available server that manages the File System
Namespace and controls access to files by clients. I will be discussing this High
Availability feature of Apache Hadoop HDFS in my next blog. The HDFS
architecture is built in such a way that the user data never resides on the
NameNode. The data resides on DataNodes only.

26
 CCS334 BIG DATA ANALYTICS

Functions of NameNode:

 It is the master daemon that maintains and manages the DataNodes (slave nodes)
 It records the metadata of all the files stored in the cluster, e.g. The location of
blocks stored, the size of the files, permissions, hierarchy, etc. There are two files
associated with the metadata:
o FsImage: It contains the complete state of the file system namespace since the
start of the NameNode.
o EditLogs: It contains all the recent modifications made to the file system with
respect to the most recent FsImage.
 It records each change that takes place to the file system metadata. For example, if
a file is deleted in HDFS, the NameNode will immediately record this in the
EditLog.
 It regularly receives a Heartbeat and a block report from all the DataNodes in the
cluster to ensure that the DataNodes are live.
 It keeps a record of all the blocks in HDFS and in which nodes these blocks are
located.
 The NameNode is also responsible to take care of the replication factor of all the
blocks which we will discuss in detail later in this HDFS tutorial blog.
 In case of the DataNode failure, the NameNode chooses new DataNodes for new
replicas, balance disk usage and manages the communication traffic to the
DataNodes.

DataNode:

DataNodes are the slave nodes in HDFS. Unlike NameNode, DataNode is a

commodity hardware, that is, a non-expensive system which is not of high quality

27
 CCS334 BIG DATA ANALYTICS

or high-availability. The DataNode is a block server that stores the data in the local
file ext3 or ext4.

Functions of DataNode:

 These are slave daemons or process which runs on each slave machine.
 The actual data is stored on DataNodes.
 The DataNodes perform the low-level read and write requests from the file
system’s clients.
 They send heartbeats to the NameNode periodically to report the overall health of
HDFS, by default, this frequency is set to 3 seconds.

Till now, you must have realized that the NameNode is pretty much important to
us. If it fails, we are doomed. But don’t worry, we will be talking about how
Hadoop solved this single point of failure problem in the next Apache Hadoop
HDFS Architecture blog. So, just relax for now and let’s take one step at a time.

Secondary NameNode:

Apart from these two daemons, there is a third daemon or a process called
Secondary NameNode. The Secondary NameNode works concurrently with the
primary NameNode as a helper daemon. And don’t be confused about the
Secondary NameNode being a backup NameNode because it is not.

28
 CCS334 BIG DATA ANALYTICS

Functions of Secondary NameNode:

 The Secondary NameNode is one which constantly reads all the file systems and
metadata from the RAM of the NameNode and writes it into the hard disk or the
file system.
 It is responsible for combining the EditLogs with FsImage from the NameNode.
 It downloads the EditLogs from the NameNode at regular intervals and applies to
FsImage. The new FsImage is copied back to the NameNode, which is used
whenever the NameNode is started the next time.

Hence, Secondary NameNode performs regular checkpoints in HDFS. Therefore, it

is also called CheckpointNode.

Blocks:

The data in HDFS is scattered across the DataNodes as blocks. Blocks are the
nothing but the smallest continuous location on your hard drive where data is
stored. In general, in any of the File System, you store the data as a collection of
blocks. Similarly, HDFS stores each file as blocks which are scattered throughout
the Apache Hadoop cluster. The default size of each block is 128 MB in Apache
Hadoop 2.x (64 MB in Apache Hadoop 1.x) which you can configure as per your
requirement.

29
 CCS334 BIG DATA ANALYTICS

It is not necessary that in HDFS, each file is stored in exact multiple of the
configured block size (128 MB, 256 MB etc.). Let’s take an example where I have
a file “[Link]” of size 514 MB as shown in above figure. Suppose that we are
using the default configuration of block size, which is 128 MB. Then, how many
blocks will be created? 5, Right. The first four blocks will be of 128 MB. But, the
last block will be of 2 MB size only.

Replication Management:

HDFS provides a reliable way to store huge data in a distributed environment as

data blocks. The blocks are also replicated to provide fault tolerance. The default
replication factor is 3 which is again configurable. So, as you can see in the figure
below where each block is replicated three times and stored on different
DataNodes (considering the default replication factor.

30
 CCS334 BIG DATA ANALYTICS

Therefore, if you are storing a file of 128 MB in HDFS using the default
configuration, you will end up occupying a space of 384 MB (3*128 MB) as the
blocks will be replicated three times and each replica will be residing on a different
DataNode.

Note: The NameNode collects block report from DataNode periodically to

maintain the replication factor. Therefore, whenever a block is over-replicated or
under-replicated the NameNode deletes or add replicas as needed.

Rack Awareness:

31
 CCS334 BIG DATA ANALYTICS

Anyways, moving ahead, let’s talk more about how HDFS places replica and what
is rack awareness? Again, the NameNode also ensures that all the replicas are not
stored on the same rack or a single rack. It follows an in-built Rack Awareness
Algorithm to reduce latency as well as provide fault tolerance. Considering the
replication factor is 3, the Rack Awareness Algorithm says that the first replica of a
block will be stored on a local rack and the next two replicas will be stored on a
different (remote) rack but, on a different DataNode within that (remote) rack as
shown in the figure above. If you have more replicas, the rest of the replicas will
be placed on random DataNodes provided not more than two replicas reside on the
same rack, if possible. You can even check out the details of Big Data with
the Azure Data Engineering Course in Singapore.

This is how an actual Hadoop production cluster looks like. Here, you have
multiple racks populated with DataNodes:

32
 CCS334 BIG DATA ANALYTICS

Advantages of Rack Awareness:

So, now you will be thinking why do we need a Rack Awareness algorithm? The
reasons are:

 To improve the network performance: The communication between nodes

residing on different racks is directed via switch. In general, you will find greater
network bandwidth between machines in the same rack than the machines residing
in different rack. So, the Rack Awareness helps you to have reduce write traffic in
between different racks and thus providing a better write performance. Also, you

33
 CCS334 BIG DATA ANALYTICS

will be gaining increased read performance because you are using the bandwidth of
multiple racks.

 To prevent loss of data: We don’t have to worry about the data even if an entire
rack fails because of the switch failure or power failure. And if you think about it,
it will make sense, as it is said that never put all your eggs in the same basket.

HDFS Read/ Write Architecture:

Now let’s talk about how the data read/write operations are performed on HDFS.
HDFS follows Write Once – Read Many Philosophy. So, you can’t edit files
already stored in HDFS. But, you can append new data by re-opening the file. Get
a better understanding of the Hadoop Clusters, nodes, and architecture from
the Hadoop Admin Training in Chennai.

HDFS Write Architecture:

Suppose a situation where an HDFS client, wants to write a file named

“[Link]” of size 248 MB.

Assume that the system block size is configured for 128 MB (default). So, the
client will be dividing the file “[Link]” into 2 blocks – one of 128 MB (Block
A) and the other of 120 MB (block B).

34
 CCS334 BIG DATA ANALYTICS

Now, the following protocol will be followed whenever the data is written into
HDFS:

 At first, the HDFS client will reach out to the NameNode for a Write Request
against the two blocks, say, Block A & Block B.
 The NameNode will then grant the client the write permission and will provide the
IP addresses of the DataNodes where the file blocks will be copied eventually.
 The selection of IP addresses of DataNodes is purely randomized based on
availability, replication factor and rack awareness that we have discussed earlier.
 Let’s say the replication factor is set to default i.e. 3. Therefore, for each block the
NameNode will be providing the client a list of (3) IP addresses of DataNodes. The
list will be unique for each block.
 Suppose, the NameNode provided following lists of IP addresses to the client:
o For Block A, list A = {IP of DataNode 1, IP of DataNode 4, IP of DataNode 6}
o For Block B, set B = {IP of DataNode 3, IP of DataNode 7, IP of DataNode 9}
 Each block will be copied in three different DataNodes to maintain the replication
factor consistent throughout the cluster.
 Now the whole data copy process will happen in three stages:

1. Set up of Pipeline
2. Data streaming and replication
3. Shutdown of Pipeline (Acknowledgement stage)

35
 CCS334 BIG DATA ANALYTICS

1. Set up of Pipeline:

Before writing the blocks, the client confirms whether the DataNodes, present in
each of the list of IPs, are ready to receive the data or not. In doing so, the client
creates a pipeline for each of the blocks by connecting the individual DataNodes in
the respective list for that block. Let us consider Block A. The list of DataNodes
provided by the NameNode is:

For Block A, list A = {IP of DataNode 1, IP of DataNode 4, IP of DataNode 6}.

So, for block A, the client will be performing the following steps to create a
pipeline:

36
 CCS334 BIG DATA ANALYTICS

 The client will choose the first DataNode in the list (DataNode IPs for Block A)
which is DataNode 1 and will establish a TCP/IP connection.
 The client will inform DataNode 1 to be ready to receive the block. It will also
provide the IPs of next two DataNodes (4 and 6) to the DataNode 1 where the
block is supposed to be replicated.
 The DataNode 1 will connect to DataNode 4. The DataNode 1 will inform
DataNode 4 to be ready to receive the block and will give it the IP of DataNode 6.
Then, DataNode 4 will tell DataNode 6 to be ready for receiving the data.
 Next, the acknowledgement of readiness will follow the reverse sequence, i.e.
From the DataNode 6 to 4 and then to 1.
 At last DataNode 1 will inform the client that all the DataNodes are ready and a
pipeline will be formed between the client, DataNode 1, 4 and 6.
 Now pipeline set up is complete and the client will finally begin the data copy or
streaming process.

2. Data Streaming:

As the pipeline has been created, the client will push the data into the pipeline.
Now, don’t forget that in HDFS, data is replicated based on replication factor. So,
here Block A will be stored to three DataNodes as the assumed replication factor is
3. Moving ahead, the client will copy the block (A) to DataNode 1 only. The
replication is always done by DataNodes sequentially.

37
 CCS334 BIG DATA ANALYTICS

So, the following steps will take place during replication:

 Once the block has been written to DataNode 1 by the client, DataNode 1 will
connect to DataNode 4.
 Then, DataNode 1 will push the block in the pipeline and data will be copied to
DataNode 4.
 Again, DataNode 4 will connect to DataNode 6 and will copy the last replica of the
block.

3. Shutdown of Pipeline or Acknowledgement stage:

Once the block has been copied into all the three DataNodes, a series of
acknowledgements will take place to ensure the client and NameNode that the data

38
 CCS334 BIG DATA ANALYTICS

has been written successfully. Then, the client will finally close the pipeline to end
the TCP session.

As shown in the figure below, the acknowledgement happens in the reverse

sequence i.e. from DataNode 6 to 4 and then to 1. Finally, the DataNode 1 will
push three acknowledgements (including its own) into the pipeline and send it to
the client. The client will inform NameNode that data has been written
successfully. The NameNode will update its metadata and the client will shut down
the pipeline.

Similarly, Block B will also be copied into the DataNodes in parallel with Block
A. So, the following things are to be noticed here:

39
 CCS334 BIG DATA ANALYTICS

 The client will copy Block A and Block B to the first DataNode simultaneously.
 Therefore, in our case, two pipelines will be formed for each of the block and all
the process discussed above will happen in parallel in these two pipelines.
 The client writes the block into the first DataNode and then the DataNodes will be
replicating the block sequentially.

As you can see in the above image, there are two pipelines formed for each block
(A and B). Following is the flow of operations that is taking place for each block in
their respective pipelines:

 For Block A: 1A -> 2A -> 3A -> 4A

 For Block B: 1B -> 2B -> 3B -> 4B -> 5B -> 6B

40
 CCS334 BIG DATA ANALYTICS

HDFS Read Architecture:

HDFS Read architecture is comparatively easy to understand. Let’s take the above
example again where the HDFS client wants to read the file “[Link]” now.

Weekday / Weekend Batche

Now, following steps will be taking place while reading the file:

 The client will reach out to NameNode asking for the block metadata for the file
“[Link]”.
 The NameNode will return the list of DataNodes where each block (Block A and
B) are stored.
 After that client, will connect to the DataNodes where the blocks are stored.
41
 CCS334 BIG DATA ANALYTICS

 The client starts reading data parallel from the DataNodes (Block A from
DataNode 1 and Block B from DataNode 3).
 Once the client gets all the required file blocks, it will combine these blocks to
form a file.

While serving read request of the client, HDFS selects the replica which is closest
to the client. This reduces the read latency and the bandwidth consumption.
Therefore, that replica is selected which resides on the same rack as the reader
node, if possible.

Now, you should have a pretty good idea about Apache Hadoop HDFS
Architecture. I understand that there is a lot of information here and it may not be
easy to get it in one go. I would suggest you to go through it again and I am sure
you will find it easier this time. Now, in my next blog, I will be talking about
Apache Hadoop HDFS Federation and High Availability Architecture.

Java Interface

Hadoop has an abstract notion of filesystems, of which HDFS is just one

implementation. The Java abstract class [Link]
represents the client interface to a filesystem in Hadoop, and there are several
concrete [Link] is written in Java, so most Hadoop filesystem
interactions are mediated through the Java API. The filesystem shell, for example,
is a Java application that uses the Java FileSystem class to provide filesystem
[Link] exposing its filesystem interface as a Java API, Hadoop makes it
awkward for non-Java applications to access HDFS. The HTTP REST API
exposed by the WebHDFS protocol makes it easier for other languages to interact

42
 CCS334 BIG DATA ANALYTICS

with HDFS. Note that the HTTP interface is slower than the native Java client, so
should be avoided for very large data transfers if possible.

There are two ways of accessing HDFS over HTTP: directly, where the HDFS
daemons serve HTTP requests to clients and via a proxy, which accesses HDFS on
the client’s behalf using the usual DistributedFileSystem API.

In the first case, the embedded web servers in the namenode and datanodes act as
WebHDFS endpoints. (WebHDFS is enabled by default, since
[Link] is set to true.) File metadata operations are handled by the
namenode, while file read (and write) operations are sent first to the namenode,
which sends an HTTP redirect to the client indicating the datanode to stream file
data from (or to).

The second way of accessing HDFS over HTTP relies on one or more standalone
proxy servers. All traffic to the cluster passes through the proxy, so the client never
accesses the namenode or datanode directly. This allows for stricter firewall and
bandwidth-limiting policies to be put in place. The HttpFS proxy exposes the same
HTTP (and HTTPS) interface as WebHDFS, so clients can access both using
webhdfs (or swebhdfs) URIs. The HttpFS proxy is started independently of the
namenode and datanode daemons, using the [Link] script, and by default listens
on a different port number 14000.

The Java Interface

In this section, we dig into the Hadoop FileSystem class: the API for interacting
with one of Hadoop’s filesystems.

43
 CCS334 BIG DATA ANALYTICS

Reading Data Using the FileSystem API

A file in a Hadoop filesystem is represented by a Hadoop Path object. FileSystem

is a general filesystem API, so the first step is to retrieve an instance for the
filesystem we want to use—HDFS, in this case. There are several static factory
methods for getting a FileSystem instance

public static FileSystem get(Configuration conf) throws IOException

public static FileSystem get(URI uri, Configuration conf) throws IOException

public static FileSystem get(URI uri, Configuration conf, String user) throws
IOException

public static LocalFileSystem getLocal(Configuration conf) throws IOException

A Configuration object encapsulates a client or server’s configuration, which is set

using configuration files read from the classpath, such as [Link]. The first
method returns the default filesystem (as specified in [Link], or the default
local filesystem if not specified there). The second uses the given URI’s scheme
and authority to determine the filesystem to use, falling back to the default
filesystem if no scheme is specified in the given URI. The third retrieves the
filesystem as the given user, which is important in the context of [Link]
fourth one retrieves a local filesystem instance.

With a FileSystem instance in hand, we invoke an open() method to get the input
stream for a [Link] first method uses a default buffer size of 4 [Link] second
one gives an option to user to specify the buffer size.

public FSDataInputStream open(Path f) throws IOException

44
 CCS334 BIG DATA ANALYTICS

public abstract FSDataInputStream open(Path f, int bufferSize) throws

IOException

FSDataInputStream

The open() method on FileSystem actually returns an FSDataInputStream rather

than a standard [Link] class. This class is a specialization of
[Link] with support for random access, so you can read from any
part of the stream:

package [Link];

public class FSDataInputStream extends DataInputStream implements

Seekable, PositionedReadable {

The Seekable interface permits seeking to a position in the file and provides a
query method for the current offset from the start of the file (getPos()) . Calling
seek() with a position that is greater than the length of the file will result in an
IOException. Unlike the skip() method of [Link], which positions the
stream at a point later than the current position, seek() can move to an arbitrary,
absolute position in the file.

public interface Seekable {

void seek(long pos) throws IOException;

long getPos() throws IOException;

45
 CCS334 BIG DATA ANALYTICS

FSDataInputStream also implements the PositionedReadable interface for reading

parts of a file at a given offset.

public interface PositionedReadable {

public int read(long position, byte[] buffer, int offset, int length) throws
IOException;

public void readFully(long position, byte[] buffer, int offset, int length) throws
IOException;

public void readFully(long position, byte[] buffer) throws IOException;

The read() method reads up to length bytes from the given position in the file into
the buffer at the given offset in the buffer. The return value is the number of bytes
actually read; callers should check this value, as it may be less than [Link]
readFully() methods will read length bytes into the buffer, unless the end of the file
is reached, in which case an EOFException is thrown.

Finally, bear in mind that calling seek() is a relatively expensive operation and
should be done sparingly. You should structure your application access patterns to
rely on streaming data (by using MapReduce, for example) rather than performing
a large number of seeks.

46
 CCS334 BIG DATA ANALYTICS

Writing Data

The FileSystem class has a number of methods for creating a file. The simplest is
the method that takes a Path object for the file to be created and returns an output
stream to write to.

public FSDataOutputStream create(Path f) throws IOException

There are overloaded versions of this method that allow you to specify whether to
forcibly overwrite existing files, the replication factor of the file, the buffer size to
use when writing the file, the block size for the file, and file permissions.

Note : The create() methods create any parent directories of the file to be written
that don’t already exist. Though convenient, this behavior may be unexpected. If
you want the write to fail when the parent directory doesn’t exist, you should check
for the existence of the parent directory first by calling the exists() method.
Alternatively, use FileContext, which allows you to control whether parent
directories are created or not.

There’s also an overloaded method for passing a callback interface, Progressable,

so your application can be notified of the progress of the data being written to the
datanodes.

package [Link];

public interface Progressable {

public void progress();

47
 CCS334 BIG DATA ANALYTICS

As an alternative to creating a new file, you can append to an existing file using the
append() method (there are also some other overloaded versions)

public FSDataOutputStream append(Path f) throws IOException

The append operation allows a single writer to modify an already written file by
opening it and writing data from the final offset in the file. With this API,
applications that produce unbounded files, such as logfiles, can write to an existing
file after having closed it. The append operation is optional and not implemented
by all Hadoop filesystems.

Below example shows how to copy a local file to a Hadoop filesystem. We

illustrate progress by printing a period every time the progress() method is called
by Hadoop, which is after each 64 KB packet of data is written to the datanode
pipeline.

import [Link];

import [Link];

import [Link];

import [Link];

import [Link];

import [Link];

import [Link];

import [Link];

48
 CCS334 BIG DATA ANALYTICS

import [Link];

import [Link];

public class CopyFileWithProgress {

public static void main(String[] args) throws Exception {

String localSrc = args[0];

String dst = args[1];

InputStream in = new BufferedInputStream(new FileInputStream(localSrc));

Configuration conf = new Configuration();

FileSystem fs = [Link]([Link](dst), conf);

OutputStream out = [Link](new Path(dst), new Progressable() {

public void progress() {

[Link](".");

});

[Link](in, out, 4096, true);

49
 CCS334 BIG DATA ANALYTICS

FSDataOutputStream

The create() method on FileSystem returns an FSDataOutputStream, which, like

FSDataInputStream, has a method for querying the current position in the file:

package [Link];

public class FSDataOutputStream extends DataOutputStream implements

Syncable {

public long getPos() throws IOException {

However, unlike FSDataInputStream, FSDataOutputStream does not permit

seeking. This is because HDFS allows only sequential writes to an open file or
appends to an already written file. In other words, there is no support for writing to
anywhere other than the end of the file, so there is no value in being able to seek
while writing.

FileSystem provides a method to create a directory also

public boolean mkdirs(Path f) throws IOException

This method creates all of the necessary parent directories if they don’t already
exists and returns true if its success full.

50
 CCS334 BIG DATA ANALYTICS

Querying the Filesystem

The FileStatus class encapsulates filesystem metadata for files and directories,
including file length, block size, replication, modification time, ownership, and
permission information. And also it gives the ability to navigate its directory
structure and retrieve information about the files and directories

The method getFileStatus() on FileSystem provides a way of getting a FileStatus

object for a single file or directory.

If no file or directory exists, a FileNotFoundException is thrown. However, if you

are interested only in the existence of a file or directory, the exists() method on
FileSystem is more convenient:

public boolean exists(Path f) throws IOException

Some examples of using the java api for hdfs

Scenario 1 – Given the namenode url,source directory and target directory copy all
the files inside the source directory to target directory.

public class HdfsFileSystem {

private static Configuration conf = new Configuration();

public static void copyfiles(String url, String sourceDirectory, String

destDirectory) throws Exception {

FileSystem fs = [Link]([Link](url), conf);

Path[] paths = getAbsolutePaths(sourceDirectory);

51
 CCS334 BIG DATA ANALYTICS

for (Path path : paths) {

if ( && [Link]().endsWith(".[Link]")) {

[Link](fs, path, fs, new Path(sourceDirectory + [Link] +

[Link]()), false, conf);

public static Path[] getAbsolutePaths(String path) throws IOException {

FileSystem fs = [Link]([Link](path), conf);

Path fsPath = new Path(path);

if ([Link](fsPath)) {

FileStatus[] status = [Link](fsPath);

return FileUtil.stat2Paths(status);

} else {

return null;

52
 CCS334 BIG DATA ANALYTICS

Scenario 2 – Given the namenode url and the path create a success trigger file with
can be used to mark successfull completion of a job.

public static void createSuccessFile(String url, String path) throws IOException

{

FileSystem fs = [Link]([Link](path), conf);

[Link](new Path(path));

Scenario 3 – Given the url to the path of the file return the content of the file in
string

public static String getFileContent(String url) throws IOException {

FileSystem fs = [Link]([Link](url), conf);

InputStream content = null;

String fileContent = null;

try {

content = [Link](new Path(url));

fileContent = [Link](content);

} finally {

53
 CCS334 BIG DATA ANALYTICS

[Link](content);

return fileContent;

Scenario 4 – renaming the filename.

public static boolean renameFile(String input, String output) throws IOException

{

FileSystem fs = [Link]([Link](input), conf);

return [Link](new Path(input), new Path(output));

Scenario 5 – Given a fileContent in the form of string create a file and write the
content of the string to the file.

public static void writeFileContent(String ouputUrl, String fileContents) throws

IOException {

FileSystem fs = [Link]([Link](ouputUrl), conf);

InputStream ins = new ByteArrayInputStream([Link]());

OutputStream os = [Link](new Path(ouputUrl));

try {

54
 CCS334 BIG DATA ANALYTICS

[Link](ins, os, 4096, false);

} finally {

[Link](ins);

[Link](os);

Scenario 6 – Given a directory location return the list of files inside the directory.

public static List<String> getFileNames(String path) throws IOException {

FileSystem fs = [Link]([Link](path), conf);

Path fsPath = new Path(path);

FileStatus[] status = [Link](fsPath);

Path[] listedPaths = FileUtil.stat2Paths(status);

List<String> files = new ArrayList<String>();

for (Path file : listedPaths) {

[Link]([Link]());

return files;

55
 CCS334 BIG DATA ANALYTICS

Scenario 7 – Given a base path and the list of file inside the base path delete all the
files in the list.

public static void deleteFiles(List<String> listOfFiles, String basePath) throws

Exception {

[Link]("Delete many files function called with base path :" +

basePath);

FileSystem fs = [Link]([Link](basePath), conf);

for (String filePath : listOfFiles) {

Path path = new Path(basePath + [Link] + filePath);

boolean test = [Link](path, false);

if (test) {

[Link]("File deleted successfull : " + basePath + [Link] +

filePath);

} else {

[Link]("File deletion failed : " + basePath + [Link] + filePath);

}
56
 CCS334 BIG DATA ANALYTICS

Scenario 8 – given a url check whether the file exists

public static boolean fileExists(String url) throws IllegalArgumentException,

IOException {

FileSystem fs = [Link]([Link](url), conf);

return [Link](new Path(url));

File patterns

It is a common requirement to process sets of files in a single operation. For

example, a MapReduce job for log processing might analyze a month’s worth of
files contained in a number of directories. Rather than having to enumerate each
file and directory to specify the input, it is convenient to use wildcard characters to
match multiple files with a single expression, an operation that is known as
globbing. Hadoop provides two FileSystem methods for processing globs:

public FileStatus[] globStatus(Path pathPattern) throws IOException

public FileStatus[] globStatus(Path pathPattern, PathFilter filter)

throws IOException

The globStatus() methods return an array of FileStatus objects whose paths match
the supplied pattern, sorted by path. An optional PathFilter can be specified to
restrict the matches further.

57
 CCS334 BIG DATA ANALYTICS

PathFilter

Glob patterns are not always powerful enough to describe a set of files you want to
access. For example, it is not generally possible to exclude a particular file using a
glob pattern. The listStatus() and globStatus() methods of FileSystem take an
optional PathFilter, which allows programmatic control over matching.

package [Link];

public interface PathFilter {

boolean accept(Path path);

PathFilter is the equivalent of [Link] for Path objects rather than File
objects. The filter passes only those files that don’t match the regular expression.
After the glob picks out an initial set of files to include, the filter is used to refine
the results

import [Link];

import [Link];

public class HdfsPathFilter implements PathFilter {

private final String regex;

public HdfsPathFilter(String regex) {

[Link] = regex;

58
 CCS334 BIG DATA ANALYTICS

public boolean accept(Path path) {

return ![Link]().matches(regex);

DATA FLOW IN HDFS

Hadoop HDFS Data Read and Write Operations

HDFS follow Write once Read many models. So we cannot edit files already
stored in HDFS, but we can append data by reopening the file. In Read-Write
operation client first, interact with the NameNode. NameNode provides privileges
so, the client can easily read and write data blocks into/from the respective
datanodes. In this blog, we will discuss the internals of Hadoop HDFS data read
and write operations. We will also cover how client read and write the data
from HDFS, how the client interacts with master and slave nodes in HDFS data
read and

Hadoop HDFS Data Write Operation

To write a file in HDFS, a client needs to interact with master

i.e. namenode (master). Now namenode provides the address of
the datanodes (slaves) on which client will start writing the data. Client directly
writes data on the datanodes, now datanode will create data write pipeline.
The first datanode will copy the block to another datanode, which intern copy it to

59
 CCS334 BIG DATA ANALYTICS

the third datanode. Once it creates the replicas of blocks, it sends back the
acknowledgment.

HDFS Data Write Pipeline Workflow

Now let’s understand complete end to end HDFS data write pipeline. As shown in
the above figure the data write operation in HDFS is distributed, client copies the
data distributedly on datanodes, the steps by step explanation of data write
operation is:

i) The HDFS client sends a create request on DistributedFileSystem APIs.

ii) DistributedFileSystem makes an RPC call to the namenode to create a new file
in the file system’s namespace.

The namenode performs various checks to make sure that the file doesn’t already
exist and that the client has the permissions to create the file. When these checks
pass, then only the namenode makes a record of the new file; otherwise, file
creation fails and the client is thrown an IOException.

iii) The DistributedFileSystem returns a FSDataOutputStream for the client to start

writing data to. As the client writes data, DFSOutputStream splits it into packets,
which it writes to an internal queue, called the data queue. The data queue is
consumed by the DataStreamer, whichI is responsible for asking the namenode to
allocate new blocks by picking a list of suitable datanodes to store the replicas.

iv) The list of datanodes form a pipeline, and here we’ll assume the replication
level is three, so there are three nodes in the pipeline. The DataStreamer streams
the packets to the first datanode in the pipeline, which stores the packet and

60
 CCS334 BIG DATA ANALYTICS

forwards it to the second datanode in the pipeline. Similarly, the second datanode
stores the packet and forwards it to the third (and last) datanode in the pipeline.

v) DFSOutputStream also maintains an internal queue of packets that are waiting

to be acknowledged by datanodes, called the ack queue. A packet is removed from
the ack queue only when it has been acknowledged by the datanodes in the
pipeline. Datanode sends the acknowledgment once required replicas are created (3
by default). Similarly, all the blocks are stored and replicated on the different
datanodes, the data blocks are copied in parallel.

vi) When the client has finished writing data, it calls close() on the stream.

vii) This action flushes all the remaining packets to the datanode pipeline and waits
for acknowledgments before contacting the namenode to signal that the file is
complete. The namenode already knows which blocks the file is made up of, so it
only has to wait for blocks to be minimally replicated before returning
successfully.

61
 CCS334 BIG DATA ANALYTICS

b. How to Write a file in HDFS – Java Program

A sample code to write a file to HDFS in Java is as follows :

FileSystem fileSystem = [Link](conf);

// Check if the file already exists
Path path = new Path(“/path/to/[Link]”);
if ([Link](path)) {
[Link](“File ” + dest + ” already exists”);
return;
}
// Create a new file and write data to it.
FSDataOutputStream out = [Link](path);
InputStream in = new BufferedInputStream(new FileInputStream(
new File(source)));
byte[] b = new byte[1024];
int numBytes = 0;
while ((numBytes = [Link](b)) > 0) {
[Link](b, 0, numBytes);
}
// Close all the file descripters
[Link]();
[Link]();
[Link]();

Hadoop HDFS Data Read Operation

62
 CCS334 BIG DATA ANALYTICS

To read a file from HDFS, a client needs to interact with namenode (master) as
namenode is the centerpiece of Hadoop cluster (it stores all the metadata i.e. data
about the data). Now namenode checks for required privileges, if the client has
sufficient privileges then namenode provides the address of the slaves where a file
is stored. Now client will interact directly with the respective datanodes to read the
data blocks.

a. HDFS File Read Workflow

Now let’s understand complete end to end HDFS data read operation. As shown in
the above figure the data read operation in HDFS is distributed, the client reads the
data parallelly from datanodes, the steps by step explanation of data read cycle is:

63
 CCS334 BIG DATA ANALYTICS

i) Client opens the file it wishes to read by calling open() on the FileSystem object,
which for HDFS is an instance of DistributedFileSystem. See Data Read
Operation in HDFS
ii) DistributedFileSystem calls the namenode using RPC to determine the locations
of the blocks for the first few blocks in the file. For each block, the namenode
returns the addresses of the datanodes that have a copy of that block and datanode
are sorted according to their proximity to the client.
iii) DistributedFileSystem returns a FSDataInputStream to the client for it to read
data from. FSDataInputStream, thus, wraps the DFSInputStream which manages
the datanode and namenode I/O. Client calls read() on the stream.
DFSInputStream which has stored the datanode addresses then connects to the
closest datanode for the first block in the file.
iv) Data is streamed from the datanode back to the client, as a result client can
call read() repeatedly on the stream. When the block ends, DFSInputStream will
close the connection to the datanode and then finds the best datanode for the next
block. Also learn about Data write operation in HDFS
v) If the DFSInputStream encounters an error while communicating with a
datanode, it will try the next closest one for that block. It will also remember
datanodes that have failed so that it doesn’t needlessly retry them for later blocks.
The DFSInputStream also verifies checksums for the data transferred to it from the
datanode. If it finds a corrupt block, it reports this to the namenode before
the DFSInputStream attempts to read a replica of the block from another
[Link]) When the client has finished reading the data, it calls close() on the
stream.

64
 CCS334 BIG DATA ANALYTICS

b. How to Read a file from HDFS – Java Program

A sample code to read a file from HDFS is as follows (To perform HDFS read and
write operations follow this HDFS command part – 3):
[php]FileSystem fileSystem = [Link](conf);
Path path = new Path(“/path/to/[Link]”);
if () {
[Link](“File does not exists”);
return;
}
FSDataInputStream in = [Link](path);
int numBytes = 0;
while ((numBytes = [Link](b))> 0) {
[Link]((char)numBytes));// code to manipulate the data which is read
}
[Link]();
[Link]();
[Link]();[/php]

HADOOP I/O

Hadoop comes with a set of primitives for data I/O. Some of these are techniques
that are more general than Hadoop, such as data integrity and compression, but
deserve special consideration when dealing with multiterabyte datasets. Others are
Hadoop tools or APIs that form the building blocks for developing distributed
systems, such as serialization frameworks and on-disk data structures.

65
 CCS334 BIG DATA ANALYTICS

Data Integrity

Users of Hadoop rightly expect that no data will be lost or corrupted during storage
or processing. However, since every I/O operation on the disk or network carries
with it a small chance of introducing errors into the data that it is reading or
writing, when the volumes of data flowing through the system are as large as the
ones Hadoop is capable of handling, the chance of data corruption occurring is
high.

The usual way of detecting corrupted data is by computing a checksum for the data
when it first enters the system, and then whenever it is transmitted across a channel
that is unreliable and hence capable of corrupting the data. The data is deemed to
be corrupt if the newly generated checksum doesn’t exactly match the original.
This technique doesn’t offer any way to fix the data—merely error detection. (And
this is a reason for not using low-end hardware; in particular, be sure to use ECC
memory.) Note that it is possible that it’s the checksum that is corrupt, not the data,
but this is very unlikely, since the checksum is much smaller than the data.

A commonly used error-detecting code is CRC-32 (cyclic redundancy check),

which computes a 32-bit integer checksum for input of any size.

Data Integrity in HDFS

HDFS transparently checksums all data written to it and by default verifies

checksums when reading data. A separate checksum is created for
every [Link] bytes of data. The default is 512 bytes, and since a
CRC-32 checksum is 4 bytes long, the storage overhead is less than 1%.

66
 CCS334 BIG DATA ANALYTICS

Datanodes are responsible for verifying the data they receive before storing the
data and its checksum. This applies to data that they receive from clients and from
other datanodes during replication. A client writing data sends it to a pipeline of
datanodes (as explained in Chapter 3), and the last datanode in the pipeline verifies
the checksum. If it detects an error, the client receives a ChecksumException, a
subclass of IOException.

When clients read data from datanodes, they verify checksums as well, comparing
them with the ones stored at the datanode. Each datanode keeps a persistent log of
checksum verifications, so it knows the last time each of its blocks was verified.
When a client successfully verifies a block, it tells the datanode, which updates its
log. Keeping statistics such as these is valuable in detecting bad disks.

Aside from block verification on client reads, each datanode runs

a DataBlockScanner in a background thread that periodically verifies all the blocks
stored on the datanode. This is to guard against corruption due to “bit rot” in the
physical storage media. See Datanode block scanner for details on how to access
the scanner reports.

Since HDFS stores replicas of blocks, it can “heal” corrupted blocks by copying
one of the good replicas to produce a new, uncorrupt replica. The way this works is
that if a client detects an error when reading a block, it reports the bad block and
the datanode it was trying to read from to the namenode before throwing
a ChecksumException. The namenode marks the block replica as corrupt, so it
doesn’t direct clients to it, or try to copy this replica to another datanode. It then
schedules a copy of the block to be replicated on another datanode, so its
replication factor is back at the expected level. Once this has happened, the corrupt
replica is deleted.

67
 CCS334 BIG DATA ANALYTICS

It is possible to disable verification of checksums by passing false to

the setVerifyChecksum() method on FileSystem, before using the open() method to
read a file. The same effect is possible from the shell by using the -
ignoreCrc option with the -get or the equivalent -copyToLocal command. This
feature is useful if you have a corrupt file that you want to inspect so you can
decide what to do with it. For example, you might want to see whether it can be
salvaged before you delete it.

LocalFileSystem

The Hadoop LocalFileSystem performs client-side checksumming. This means that

when you write a file called filename, the filesystem client transparently creates a
hidden file, .[Link], in the same directory containing the checksums for each
chunk of the file. Like HDFS, the chunk size is controlled by
the [Link] property, which defaults to 512 bytes. The chunk size is
stored as metadata in the .crc file, so the file can be read back correctly even if the
setting for the chunk size has changed. Checksums are verified when the file is
read, and if an error is detected, LocalFileSystem throws a ChecksumException.

Checksums are fairly cheap to compute (in Java, they are implemented in native
code), typically adding a few percent overhead to the time to read or write a file.
For most applications, this is an acceptable price to pay for data integrity. It is,
however, possible to disable checksums: the use case here is when the underlying
filesystem support checksums natively. This is accomplished by
using RawLocalFileSystem in place of LocalFileSystem. To do this globally in an
application, it suffices to remap the implementation for file URIs by setting the
property [Link] to the value [Link].
Alternatively, you can directly create a RawLocalFileSystem instance, which may

68
 CCS334 BIG DATA ANALYTICS

be useful if you want to disable checksum verification for only some reads; for
example:

Configuration conf = ...

FileSystem fs = new RawLocalFileSystem();

[Link](null, conf);

ChecksumFileSystem

LocalFileSystem uses ChecksumFileSystem to do its work, and this class makes it

easy to add checksumming to other (nonchecksummed) filesystems,
as ChecksumFileSystem is just a wrapper around FileSystem. The general idiom is
as follows:

FileSystem rawFs = ...

FileSystem checksummedFs = new ChecksumFileSystem(rawFs);

The underlying filesystem is called the raw filesystem, and may be retrieved using
the getRawFileSystem() method
on ChecksumFileSystem. ChecksumFileSystem has a few more useful methods for
working with checksums, such as getChecksumFile() for getting the path of a
checksum file for any file. Check the documentation for the others.

If an error is detected by ChecksumFileSystem when reading a file, it will call

its reportChecksumFailure() method. The default implementation does nothing,

69
 CCS334 BIG DATA ANALYTICS

but LocalFileSystem moves the offending file and its checksum to a side directory
on the same device called bad_files. Administrators should periodically check for
these bad files and take action on them.

Data Compression in Hadoop

Compression, in simple terms, refers to the process of reducing the size (number of
bits) of a file or data stream. The primary goal of compression is to store or
transmit data in a more efficient manner, ultimately saving storage space, reducing
bandwidth requirements, and improving overall system performance.

There are two types of compression techniques: lossless compression

and lossy compression.

Lossless Compression

In lossless compression, the compressed data can be fully recovered to its original
form without any loss of information. It achieves this by identifying and removing
redundant or repetitive patterns within the data. Common lossless compression
algorithms include Gzip and Deflate. Lossless compression is used when it is
crucial to retain the accuracy and integrity of the data.

Lossy Compression

Lossy compression, on the other hand, sacrifices some degree of data accuracy for
more significant compression ratios. It achieves higher compression by discarding
less essential or perceptually insignificant data. Lossy compression techniques are
widely used in multimedia applications, such as image and video compression.
Popular algorithms like JPEG and MP3 employ lossy compression to achieve
significant file size reductions.
70
 CCS334 BIG DATA ANALYTICS

Data compression at various stages in Hadoop

You can compress data in Hadoop MapReduce at various stages.

1. Compressing input files- You can compress the input file that will reduce
storage space in HDFS. If you compress the input files then the files will be
decompressed automatically when the file is processed by a MapReduce job.
Determining the appropriate coded will be done using the file name extension. As
example if file name extension is .snappy hadoop framework will automatically
use SnappyCodec to decompress the file.
2. Compressing the map output- You can compress the intermediate map
output. Since map output is written to disk and data from several map outputs is
used by a reducer so data from map outputs is transferred to the node where reduce
task is running. Thus by compressing intermediate map output you can reduce both
storage space and data transfer across network.
3. Compressing output files- You can also compress the output of a
MapReduce job.

Hadoop compression formats

There are many different compression formats available in Hadoop framework.

You will have to use one that suits your requirement.

Parameters that you need to look for are-

1. Time it takes to compress.

2. Space saving.
3. Compression format is splittable or not.

71
 CCS334 BIG DATA ANALYTICS

Let’s go through the list of available compression formats and see which format
provides what characteristics.

Deflate– It is the compression algorithm whose implementation is zlib. Defalte

compression algorithm is also used by gzip compression tool. Filename extension
is .deflate.

gzip- gzip compression is based on Deflate compression algorithm. Gzip

compression is not as fast as LZO or snappy but compresses better so space saving
is more.
Gzip is not splittable.
Filename extension is .gz.

bzip2- Using bzip2 for compression will provide higher compression ratio but the
compressing and decompressing speed is slow. Bzip2 is splitable, Bzip2Codec
implements Splitable Compression Codec interface which provides the capability
to compress / de-compress a stream starting at any arbitrary position.
Filename extension is .bz2.

Snappy– The Snappy compressor from Google provides fast compression and
decompression but compression ratio is less.
Snappy is not splittable.
Filename extension is .snappy.

LZO– LZO, just like snappy is optimized for speed so compresses and
decompresses faster but compression ratio is less.
LZO is not splittable by default but you can index the lzo files as a pre-processing
step to make them splittable.
Filename extension is .lzo.

72
 CCS334 BIG DATA ANALYTICS

LZ4– Has fast compression and decompression speed but compression ratio is
less. LZ4 is not splittable.
Filename extension is .lz4.

Zstandard– Zstandard is a real-time compression algorithm,

providing high compression ratios. It offers a very wide range of
compression / speed trade -off.
Zstandard is not splittable.
Filename extension is .zstd.

Codecs in Hadoop

Codec, short form of compressor -decompressor is the implementation of

a compression-decompression algorithm. In Hadoop framework there are
different codec classes for different compression formats, you will use
the codec class for the compression format you are using. The codec
classes in Hadoop are as follows-

Deflate – [Link] or
[Link] (DeflateCodec is an alias
for DefaultCodec). This codec uses zlib compression.

Gzip – [Link]

Bzip2 – [Link].Bzip2Codec

Snappy – [Link]

LZO – [Link],
[Link]

73
 CCS334 BIG DATA ANALYTICS

LZO libraries are GPL licensed and doesn't come with Hadoop release.
Hadoop codec for LZO has to be downloaded separately.

LZ4– [Link].Lz4Codec

Zstandard – [Link] odec

Compression and Input splits

As you must be knowing MapReduce job calculates the number of input

splits for the job and as many map tasks are launched as the count of
splits. These map tasks process the data referred by input splits in
parallel.

If you compress the input file using the compression format that is not
splittable, then it won't be possible to read data at an arbitrary point in
the stream. So the map tasks won't be able to read split data. In this
scenario MapReduce won’t create input splits and whole file will be
processed by one mapper which means no advantage of parallel
processing and data transfer overhead too.

Let's try to clarify it with an example. If you have a 1 GB file it will be

partitioned and stored as 8 data blocks in HDFS (Block size is 128 MB).
MapReduce job using this file will also create 8 input splits and launch 8
mappers to process these splits in parallel.

Now, if you compress this 1 GB file using gzip (which is not splittable)
then HDFS still stores the file as 8 separate blocks. As it is not possible
to read data at an arbitrary point in the compressed gzip stream,
MapReduce job won’t calculate input splits and launch only one map task
74
 CCS334 BIG DATA ANALYTICS

to process all the 8 HDFS blocks. So you lose the advantage of parallel
processing and there is no data locality optimization too. Even if map
task is launched on the node where data for one of the block is stored
data for all the other blocks has to be transferred to the node where map
task is launched.

Here note that compression used is Splittable or not is a factor for text
files only. If you are using container file format like sequence file or
Avro then splitting is supported even if the compressor used is not
splittable like Snappy or Gzip.

Compression increases CPU processing

There is a performance cost associated with compressing and

decompressing the data. Though you save on storage and I/O activity is
less but compression and decompression requires extra CPU cycles.

Though in most of the cases compressing data increases the overall job
performance, ensure that you weigh the pros and cons and compare the
performance gains with compressed data.

Serialization

Serialization is the process of turning structured objects into a byte stream for
transmission over a network or for writing to persistent storage. Deserialization is
the process of turning a byte stream back into a series of structured objects.

75
 CCS334 BIG DATA ANALYTICS

Serialization appears in two quite distinct areas of distributed data processing: for
interprocess communication and for persistent storage.

In Hadoop, interprocess communication between nodes in the system is

implemented using remote procedure calls (RPCs). The RPC protocol uses
serialization to render the message into a binary stream to be sent to the remote
node, which then deserializes the binary stream into the original message. In
general, it is desirable that an RPC serialization format is:

Compact

A compact format makes the best use of network bandwidth, which is the most
scarce resource in a data center.

Fast

Interprocess communication forms the backbone for a distributed system, so it is

essential that there is as little performance overhead as possible for the serialization
and deserialization process.

Extensible

Protocols change over time to meet new requirements, so it should be

straightforward to evolve the protocol in a controlled manner for clients and
servers. For example, it should be possible to add a new argument to a method call,
and have the new servers accept messages in the old format (without the new
argument) from old clients.

Interoperable

76
 CCS334 BIG DATA ANALYTICS

For some systems, it is desirable to be able to support clients that are written in
different languages to the server, so the format needs to be designed to make this
possible.

On the face of it, the data format chosen for persistent storage would have different
requirements from a serialization framework. After all, the lifespan of an RPC is
less than a second, whereas persistent data may be read years after it was written.
As it turns out, the four desirable properties of an RPC’s serialization format are
also crucial for a persistent storage format. We want the storage format to be
compact (to make efficient use of storage space), fast (so the overhead in reading
or writing terabytes of data is minimal), extensible (so we can transparently read
data written in an older format), and interoperable (so we can read or write
persistent data using different languages).

Hadoop uses its own serialization format, Writables, which is certainly compact
and fast (but not so easy to extend, or use from languages other than Java). Since
Writables are central to Hadoop (MapReduce programs use them for their key and
value types), we look at them in some depth in the next section, before briefly
turning to other well-known serialization frameworks, like Apache Thrift and
Google Protocol Buffers.

The Writable Interface

The Writable interface defines two methods: one for writing its state to
a DataOutput binary stream, and one for reading its state from a DataInput binary
stream: package [Link];

77
 CCS334 BIG DATA ANALYTICS

import [Link];

import [Link];

import [Link];

public interface Writable {

void write(DataOutput out) throws IOException;

void readFields(DataInput in) throws IOException;

Writable Classes

Hadoop comes with a large selection of Writable classes in

the [Link] package. They form the class hierarchy shown
in Figure 4-1.

Writable wrappers for Java primitives

There are Writable wrappers for all the Java primitive types (see Table 4-6)
except short and char (both of which can be stored in an IntWritable). All have
a get() and a set() method for retrieving and storing the wrapped value.

Table 4-6. Writable wrapper classes for Java primitives

78
 CCS334 BIG DATA ANALYTICS

Java primitive Writable implementation

boolean BooleanWritable

byte ByteWritable

int IntWritable

VIntWritable

float FloatWritable

long LongWritable

VLongWritable

double DoubleWritable

79
 CCS334 BIG DATA ANALYTICS

Figure 4-1. Writable class hierarchy

To transfer data over a network or for its persistent storage, you need to serialize
the data. Prior to the serialization APIs provided by Java and Hadoop, we have a
special utility, called Avro, a schema-based serialization technique.

80
 CCS334 BIG DATA ANALYTICS

This tutorial teaches you how to serialize and deserialize the data using Avro. Avro
provides libraries for various programming languages. In this tutorial, we
demonstrate the examples using Java library.

What is Avro?

Apache Avro is a language-neutral data serialization system. It was developed by

Doug Cutting, the father of Hadoop. Since Hadoop writable classes lack language
portability, Avro becomes quite helpful, as it deals with data formats that can be
processed by multiple languages. Avro is a preferred tool to serialize data in
Hadoop.

Avro has a schema-based system. A language-independent schema is associated

with its read and write operations. Avro serializes the data which has a built-in
schema. Avro serializes the data into a compact binary format, which can be
deserialized by any application.

Avro uses JSON format to declare the data structures. Presently, it supports
languages such as Java, C, C++, C#, Python, and Ruby.

Avro Schemas

Avro depends heavily on its schema. It allows every data to be written with no
prior knowledge of the schema. It serializes fast and the resulting serialized data is
lesser in size. Schema is stored along with the Avro data in a file for any further
processing.

In RPC, the client and the server exchange schemas during the connection. This
exchange helps in the communication between same named fields, missing fields,
extra fields, etc.
81
 CCS334 BIG DATA ANALYTICS

Avro schemas are defined with JSON that simplifies its implementation in
languages with JSON libraries.

Like Avro, there are other serialization mechanisms in Hadoop such as Sequence
Files, Protocol Buffers, and Thrift.

Comparison with Thrift and Protocol Buffers

Thrift and Protocol Buffers are the most competent libraries with Avro. Avro
differs from these frameworks in the following ways −

 Avro supports both dynamic and static types as per the requirement. Protocol
Buffers and Thrift use Interface Definition Languages (IDLs) to specify schemas
and their types. These IDLs are used to generate code for serialization and
deserialization.
 Avro is built in the Hadoop ecosystem. Thrift and Protocol Buffers are not built in
Hadoop ecosystem.

Unlike Thrift and Protocol Buffer, Avro's schema definition is in JSON and not in
any proprietary IDL.

Property Avro Thrift & Protocol Buffer

Dynamic schema Yes No

Built into Hadoop Yes No

Schema in JSON Yes No

No need to compile Yes No

82
 CCS334 BIG DATA ANALYTICS

No need to declare IDs Yes No

Bleeding edge Yes No

Features of Avro

Listed below are some of the prominent features of Avro −

 Avro is a language-neutral data serialization system.

 It can be processed by many languages (currently C, C++, C#, Java, Python, and
Ruby).
 Avro creates binary structured format that is both compressible and splittable.
Hence it can be efficiently used as the input to Hadoop MapReduce jobs.
 Avro provides rich data structures. For example, you can create a record that
contains an array, an enumerated type, and a sub record. These datatypes can be
created in any language, can be processed in Hadoop, and the results can be fed to
a third language.
 Avro schemas defined in JSON, facilitate implementation in the languages that
already have JSON libraries.
 Avro creates a self-describing file named Avro Data File, in which it stores data
along with its schema in the metadata section.
 Avro is also used in Remote Procedure Calls (RPCs). During RPC, client and
server exchange schemas in the connection handshake.

General Working of Avro

To use Avro, you need to follow the given workflow −

83
 CCS334 BIG DATA ANALYTICS

 Step 1 − Create schemas. Here you need to design Avro schema according to your
data.
 Step 2 − Read the schemas into your program. It is done in two ways −
o By Generating a Class Corresponding to Schema − Compile the schema using
Avro. This generates a class file corresponding to the schema
o By Using Parsers Library − You can directly read the schema using parsers
library.
 Step 3 − Serialize the data using the serialization API provided for Avro, which is
found in the package [Link].
 Step 4 − Deserialize the data using deserialization API provided for Avro, which is
found in the package [Link].

Integrate Cassandra with Hadoop

As a data scientist you know that managing and processing large amounts of data
is no easy task Thats why many organizations turn to distributed systems like
Hadoop and Cassandra to handle their big data needs Hadoop is an opensource
framework that provides distributed storage and processing of large data sets while
Cassandra is a distributed NoSQL database that offers high scalability and
availability Integrating these two powerful technologies can provide even greater
benefits to organizations looking to manage their big data efficiently In this post
well explor As a data scientist, you know that managing and processing large
amounts of data is no easy task. That’s why many organizations turn to distributed
systems like Hadoop and Cassandra to handle their big data needs. Hadoop is an
open-source framework that provides distributed storage and processing of large
data sets, while Cassandra is a distributed NoSQL database that offers high

84
 CCS334 BIG DATA ANALYTICS

scalability and availability. Integrating these two powerful technologies can

provide even greater benefits to organizations looking to manage their big data
efficiently. In this post, we’ll explore how to integrate Cassandra with Hadoop and
the benefits of doing so.

Why Integrate Cassandra with Hadoop?

Integrating Cassandra with Hadoop provides several benefits, including:

1. Scalability: By integrating Cassandra with Hadoop, you can scale your data
processing and storage capabilities to handle even larger data sets. Cassandra’s
distributed architecture allows it to store and process data across multiple nodes,
while Hadoop’s distributed computing framework enables parallel processing of
large data sets.
2. High Availability: Cassandra’s distributed architecture also provides high
availability, ensuring that your data remains accessible even in the event of a single
node failure. This is especially important for organizations that require continuous
availability of their data.
3. Efficient Data Processing: Hadoop’s MapReduce framework allows for efficient
processing of large data sets. By integrating Cassandra with Hadoop, you can
leverage Hadoop’s processing capabilities to analyze and process data stored in
Cassandra.

How to Integrate Cassandra with Hadoop

85
 CCS334 BIG DATA ANALYTICS

Integrating Cassandra with Hadoop involves several steps, including:

Step 1: Install Hadoop and Cassandra

To integrate Cassandra with Hadoop, you need to have both technologies installed
on your system. You can download the latest versions of Hadoop and Cassandra
from their respective websites. Once downloaded, follow the installation
instructions provided by each technology.

86
 CCS334 BIG DATA ANALYTICS

Step 2: Configure Hadoop

Once you have installed Hadoop, you need to configure it to work with Cassandra.
This involves modifying the Hadoop configuration files to include the necessary
settings for connecting to Cassandra.

First, navigate to the Hadoop installation directory and open the [Link] file.
Add the following lines to the file:

<property>

<name>[Link]</name>

<value>localhost</value>

</property>

Next, open the [Link] file and add the following lines:

<property>

<name>[Link]</name>

<value>true</value>

</property>

<property>

<name>[Link]</name>

<value>localhost</value>

87
 CCS334 BIG DATA ANALYTICS

</property>

These settings configure Hadoop to use the Cassandra input and output formats.

Step 3: Configure Cassandra

Next, you need to configure Cassandra to work with Hadoop. This involves
modifying the Cassandra configuration files to include the necessary settings for
connecting to Hadoop.

First, navigate to the Cassandra installation directory and open

the [Link] file. Add the following lines to the file:

hadoop_config:

[Link]: hdfs://localhost:9000

This setting configures Cassandra to use the Hadoop file system.

Step 4: Create a Hadoop Job to Access Cassandra Data

Once you have configured Hadoop and Cassandra to work together, you can create
a Hadoop job to access the data stored in Cassandra. This involves writing a
MapReduce program that uses the Cassandra input and output formats to read and
write data.

Here’s an example MapReduce program that reads data from a Cassandra table and
writes it to a Hadoop file:

public class CassandraHadoopJob {

public static void main(String[] args) throws Exception {

88
 CCS334 BIG DATA ANALYTICS

Configuration conf = new Configuration();

Job job = [Link](conf, "Cassandra Hadoop Job");

[Link]([Link]);

[Link]([Link]);

[Link]([Link]);

[Link]([Link]);

[Link]([Link]);

[Link]([Link]);

[Link]([Link]);

[Link]([Link]);

[Link]([Link]);

[Link]([Link](),
"keyspace", "table");

[Link]([Link](),
"keyspace", "table");

[Link]([Link](),
"localhost");

[Link]([Link](), "9160");

89
 CCS334 BIG DATA ANALYTICS

[Link](job, new Path("output"));

[Link]([Link](true) ? 0 : 1);

This program uses the CassandraInputFormat and TextOutputFormat classes to

read and write data, respectively.
The CassandraMapper and CassandraReducer classes define the Map and Reduce
functions, respectively. The CassandraConfigHelper class is used to configure the
input and output column families, as well as the initial address and RPC port for
connecting to Cassandra.

Step 5: Run the Hadoop Job

Once you have written your MapReduce program, you can run the Hadoop job
using the following command:

$ hadoop jar <path-to-jar-file> <main-class> <input-path> <output-path>

Replace <path-to-jar-file> with the path to your MapReduce program’s JAR

file, <main-class> with the fully qualified name of your program’s main
class, <input-path> with the path to the input data, and <output-path> with the path
to the output data.

Integrating Cassandra with Hadoop provides several benefits to organizations

looking to manage their big data efficiently. By leveraging the scalability and
availability of Cassandra and the efficient processing capabilities of Hadoop,

90
 CCS334 BIG DATA ANALYTICS

organizations can handle even larger data sets with ease. Integrating the two
technologies involves several steps, including installing and configuring Hadoop
and Cassandra, creating a Hadoop job to access Cassandra data, and running the
job. With this guide, you’ll be able to integrate Cassandra with Hadoop and take
advantage of the benefits that come with it.

e how to integrate Cassandra with Hadoop and the benefits of doing so

91