Big Data 1

UNIT-V

MapReduce Working
Objective:
To familiarize with the working of Map Reduce in Hadoop.
Syllabus:
MapReduce Working
Classic MapReduce, Job submission, Job Initialization, Task Assignment, Task
execution, Progress and status

updates, Job completion, Shuffle and sort on Map and Reduce side, Configuration
tuning, Map Reduce types,

Input formats, Output formats.

Learning Outcomes:
At the end of the unit, students will be able to:

1. Write more advanced Map Reduce programs.

2. Describe data types supported by MapReduce and Input and Output
formats.
Learning Material
Introduction
• Run a MapReduce job with a single method call: submit() on a Job object
which will submit the job and call waitForCompletion(), wait for it to
finish.
• The steps Hadoop takes to run a job. We saw in previous chapter that the
way Hadoop executes a MapReduce program depends on a couple of
configuration settings.
• In releases of Hadoop up to and including the 0.20 release series,
[Link] determines the means of execution.

1) If this configuration property is set to local, the default, then the local job
runner is used. This runner runs the whole job in a single JVM.
IV-II SEMESTER 2018-19 CSE
 Big Data 2

2)It’s designed for testing and for running MapReduce programs on small
datasets.
3) If [Link] is set to a colon-separated host and port pair, then
the property is interpreted as a jobtracker address, and the runner
submits the job to the jobtracker at that address.
5.1 Classic MapReduce (MapReduce 1)

A job run in classic MapReduce is illustrated in Figure .

At the highest level, there are four independent entities:

1. The client, which submits the MapReduce job.

2. The jobtracker, which coordinates the job run. The jobtracker is a Java
application whose main class is JobTracker.

3. The tasktrackers, which run the tasks that the job has been split
into. Tasktrackers are Java applications whose main class is
TaskTracker.

4. The distributed filesystem (normally HDFS), which is used for sharing

job files between the other entities.

Fig : How Hadoop runs a MapReduce job using the classic framework
IV-II SEMESTER 2018-19 CSE
 Big Data 3

There are six detailed levels in workflows. They are:

1. Job Submission

2. Job Initialization

3. Task Assignment

4. Task Execution

5. Task Progress and status updates

6. Task Completion
5.2 Job Submission
The submit() method on Job creates an internal JobSummitter instance and
calls submitJobInternal() on it (step 1 in Figure).

After submitted the job, waitForCompletion() polls the job’s progress once a
second and reports the progress to the console.

When the job is complete, if it was successful, the job counters are
displayed. Otherwise, the error that caused the job to fail is logged to the
console.

The job submission process implemented by JobSummitter does the

following:

▪ Asks the jobtracker for a new job ID (by calling getNewJobId() on obTracker)
(step 2).

IV-II SEMESTER 2018-19 CSE

 Big Data 4

▪ Checks the output specification of the job. For example, if the output
directory has not been specified or it already exists, the job is not submitted
and an error is thrown to the MapReduce program.

▪ Computes the input splits for the job. If the splits cannot be computed,
because the input paths don’t exist, for example, then the job is not
submitted and an error is thrown to the MapReduce program.

▪ Copies the resources needed to run the job, including the job JAR file, the
configuration file, and the computed input splits, to the jobtracker’s
filesystem in a directory named after the job ID.

▪ The job JAR is copied with a high replication factor (controlled by the
[Link] property, which defaults to 10) so that there are
lots of copies across the cluster for the tasktrackers to access when they
run tasks for the job (step 3).

▪ Tells the jobtracker that the job is ready for execution (by calling
submitJob() on JobTracker) (step 4).

5.3 Job Initialization

• When the JobTracker receives a call to its submitJob() method, it puts it

into an internal queue from where the job scheduler will pick it up and
initialize it

Initialization involves

Bookkeeping information to keep track of the tasks’ status and progress

(step 5).

creating an object to represent the job being run, which encapsulates its
tasks, and

IV-II SEMESTER 2018-19 CSE

 Big Data 5

1. To create the list of tasks to run, the job scheduler first retrieves the input
splits computed by the client from the shared filesystem (step 6).

2. It then creates one map task for each split.

3. The number of reduce tasks to create is determined by the

[Link] property in the Job, which is set by the
setNumReduceTasks() method, and the scheduler simply creates this
number of reduce tasks to be run.

4. Tasks are given IDs at this point. In addition to the map and reduce tasks,
two further tasks are created: a job setup task and a job cleanup task.

5. These are run by tasktrackers and are used to run code to setup the job
before any map tasks run, and to clean up after all the reduce tasks are
complete.

6. The OutputCommitter that is configured for the job determines the code to
be run, andby default this is a FileOutputCommitter.

7. For the job setup task it will create the final output directory for the job and
the temporary working space for the task output, and for the job cleanup
task it will delete the temporary working space for the task output.
5.4 Task Assignment
• Tasktrackers run a simple loop that periodically sends heartbeat method
calls to the jobtracker.

• Heartbeats tell the jobtracker that a tasktracker is alive, but they also
double as a channel for messages.

• As a part of the heartbeat, a tasktracker will indicate whether it is ready to

run a new task, and if it is, the jobtracker will allocate it a task, which it
communicates to the tasktracker using the heartbeat return value (step 7).

IV-II SEMESTER 2018-19 CSE

 Big Data 6

• Before it can choose a task for the tasktracker, the jobtracker must choose a
job to select the task from. There are various scheduling algorithms but the
default one simply maintains a priority list of jobs.

• Having chosen a job, the jobtracker now chooses a task for the job.
Tasktrackers have a fixed number of slots for map tasks and for reduce
tasks:

Ex: a tasktracker may be able to run two map tasks and two reduce tasks
simultaneously. (The precise number depends on the number of cores and
the amount of memory on the tasktracker)

• The default scheduler fills empty map task slots before reduce task slots, so
if the tasktracker has at least on empty map task slot, the jobtracker will
select a map task; otherwise, it will select a reduce task.

• To choose a reduce task, the jobtracker simply takes the next in its list of
yet-to-be-run reduce tasks, since there are no data locality considerations.

• For a map task, however, it takes account of the tasktracker’s network

location and picks a task whose input split is as close as possible to the
tasktracker.

• In the optimal case, the task is data-local, that is, running on the same node
that the split resides on.

• Alternatively, the task may be rack-local: on the same rack, but not the same
node, as the split. Some tasks are neither data-local nor rack-local and
retrieve their data from a different rack from the one they are running on.

• You can tell the proportion of each type of task by looking at a job’s
counters.

IV-II SEMESTER 2018-19 CSE

 Big Data 7

5.5 Task Execution

• Now that the tasktracker has been assigned a task, the next step is for it to
run the task.

• First, it localizes the job JAR by copying it from the shared filesystem to the
tasktracker’s filesystem.

It also copies any files needed from the distributed cache by the application
to the local disk.

• Second, it creates a local working directory for the task, and un-jars the
contents of the JAR into this directory.

• Third, it creates an instance of TaskRunner to run the task. TaskRunner

launches a new Java Virtual Machine (step 9) to run each task in (step 10),
so that any bugs in the user-defined map and reduce functions don’t affect
the tasktracker (by causing it to crash or hang, for example).

• It is, however, possible to reuse the JVM between tasks.

• The child process communicates with its parent through the umbilical
interface. This way it informs the parent of the task’s progress every few
seconds until the task is complete.

• Each task can perform setup and cleanup actions, which are run in the same
JVM as the task itself, and are determined by the OutputCommitter for the
job

• The cleanup action is used to commit the task, which in the case of file-based
jobs means that its output is written to the final location for that task.

• The commit protocol ensures that when speculative execution is enabled, only
one of the duplicate tasks is committed and the other is aborted.

IV-II SEMESTER 2018-19 CSE

 Big Data 8

Fig : The relationship of the Streaming and Pipes executable to the tasktracker
and its child
• Both Streaming and Pipes run special map and reduce tasks for the purpose
of launching the user-supplied executable and communicating with it (Fig).

• In the case of Streaming, the Streaming task communicates with the process
(which may be written in any language) using standard input and output
streams.

• The Pipes task, on the other hand, listens on a socket and passes the C++
process a port number in its environment, so that on startup, the C++
process can establish a persistent socket connection back to the parent Java
Pipes task.

• In both cases, during execution of the task, the Java process passes input
key-value pairs to the external process, which runs it through the user-
defined map or reduce function and passes the output key-value pairs back
to the Java process.

• From the tasktracker’s point of view, it is as if the tasktracker child process

ran the map or reduce code itself.

IV-II SEMESTER 2018-19 CSE

 Big Data 9

5.6 Progress and Status Updates

• MapReduce jobs are long-running batch jobs, taking anything from minutes
to hours to run.

• Progress reporting is important for the user to get feedback on how the job
is progressing.

• The following operations constitute progress:

• Reading an input record ( in a mapper and reducer)

• Writing and output record ( in a mapper and reducer)

• Setting the status description on a reporter ( by using Reporter’s

setStatus() method)

• Incrementing a counter (using Reporter’s incrCounter() method)

• Calling Reporter’s Progress() method

• A job and each of its tasks have a status, which includes

• State of the job or task (e.g., running, successfully completed, failed)

• The progress of maps and reduces

• The values of the job’s counters and

• A status message or description (which may be set by user code).

• These statuses change over the course of the job, they get communicated
back to the client regarding progress by displaying

IV-II SEMESTER 2018-19 CSE

 Big Data 10

• The proportion of the task completed.

• For map tasks, the proportion of the input that has been processed.

• For reduce tasks, the proportion of the reduce input processed.

It does this by dividing the total progress into three parts,

corresponding to the three phases of the shuffle.

Display counters that count various events as the task runs

such as number of map output records written.

• If a task reports progress, it sets a flag to indicate that the status change
should be sent to the tasktracker.

• The flag is checked in a separate thread every three seconds, and if set it
notifies the tasktracker of the current task status.

• Meanwhile, the tasktracker is sending heartbeats to the jobtracker every five

seconds (this is a minimum, as the heartbeat interval is actually dependent
on the size of the cluster: for larger clusters,the interval is longer)

• The status of all the tasks being run by the tasktracker is sent in the call.

• Counters are sent less frequently than every five seconds, because they can
be relatively high-bandwidth.

• The jobtracker combines these updates to produce a global view of the

status of all the jobs being run and their constituent tasks.

• Finally, the Job receives the latest status by polling the jobtracker every
second.

IV-II SEMESTER 2018-19 CSE

 Big Data 11

• Clients can also use Job’s getStatus() method to obtain a JobStatus

instance, which contains all of the status information for the job.

5.7 Job Completion

• When the jobtracker receives a notification that the last task for a job is
complete it changes the status for the job to “successful.”

• Then, when the Job polls for status, it learns that the job has completed
successfully, it prints a message to tell the user and then returns from the
waitForCompletion() method.

• The jobtracker also sends an HTTP job notification.

• Last, the jobtracker cleans up its working state for the job and instructs
tasktrackers to do the same (so intermediate output is deleted, for example).

5.8 Shuffle and Sort

• MapReduce makes the guarantee that the input to every reducer is sorted
by key.

• The process by which the system performs the sort—and transfers the map
outputs to the reducers as inputs—is known as the shuffle.

• The shuffle is an area of the codebase where refinements and improvements

are continually being made.

Shuffle sort on Map Side

• When the map function starts producing output, it is not simply written to
disk. The process is more involved, and takes advantage of buffering writes
in memory and doing some presorting for efficiency reasons.

IV-II SEMESTER 2018-19 CSE

 Big Data 12

Shuffle and sort in MapReduce

• The buffer is 100 MB by default, change the size by using [Link]

property.

• When the contents of the buffer reaches a certain threshold size a

background thread will start to spill the contents to disk.

• Map outputs will continue to be written to the buffer while the spill takes
place, but if the buffer fills up during this time, the map will block until the
spill is complete.

• Spills are written in round-robin fashion to the directories specified by the

[Link] property.

• Before it writes to disk, the thread first divides the data into partitions
corresponding to the reducers to send.

• Within each partition, the background thread performs an in-memory sort

by key, and if there is a combiner function, it is run on the output of the
sort.

• Running the combiner function makes more compact map output, so less
data to write to local disk and to transfer to the reducer.

• Each time the memory buffer reaches the spill threshold, a new spill file is
created

• Before the task is finished, the spill files are merged into a single partitioned
and sorted output file.

• The configuration property [Link] controls the maximum number of

streams to merge at once; the default is 10.

IV-II SEMESTER 2018-19 CSE

 Big Data 13

• If there are at least three spill files then the combiner is run again before the
output file is written.

• Compress the map output as it is written to disk, makes it faster to write to

disk, saves disk space, and reduces the amount of data to transfer to the
reducer.

• By default, the output is not compressed, but it is easy to enable by setting

[Link] to true.

• The output file’s partitions are made available to the reducers over HTTP.

• The maximum number of worker threads used to serve the file partitions is
controlled by the [Link] property. The default of 40 may
need increasing for large clusters running large jobs.
Shuffle and sort on Reduce Side
The map output file is sitting on the local disk of the machine that ran the map
task. The reduce task needs the map output for its particular partition from
several map tasks across the cluster.

There are three phases for reducer 1) copy phase 2) sort phase 3) reduce phase.
1)Copy phase:

• The map tasks may finish at different times, so the reduce task starts
copying their outputs as soon as each completes.

• The reduce task has a small number of copier threads so that it can fetch
map outputs in parallel.

• The default is five threads, but this number can be changed by setting
the [Link] property.

IV-II SEMESTER 2018-19 CSE

 Big Data 14

• The map outputs are copied to reduce task JVM’s memory otherwise,
they are copied to disk.

When the in-memory buffer reaches a threshold size or reaches a

threshold number of map outputs it is merged and spilled to disk.

• Any map outputs that were compressed have to be decompressed in

memory in order to perform a merge on them.

• When all the map outputs have been copied, the reduce task moves into
the sort phase.
2)Sort phase:
• In this phase merge the map outputs, maintaining their sort
ordering.

• This is done in rounds. For example, if there were 50 map outputs,

and the merge factor was 10, then there would be 5 rounds. Each
round would merge 10 files into one, so at the end there would be five
intermediate files.

• These five files into a single sorted file, the merge saves a trip to disk
by directly feeding the reduce function. This final merge can come
from a mixture of in-memory and on-disk segments.
3)Reduce phase:
• During the reduce phase, the reduce function is invoked for each key
in the sorted output.

• The output of this phase is written directly to the output filesystem,

typically HDFS.

• In the case of HDFS, since the tasktracker node is also running a

datanode, the first block replica will be written to the local disk.

IV-II SEMESTER 2018-19 CSE

 Big Data 15

5.9 Configuration Tuning

• Configuration tuning is to tune the shuffle to improve MapReduce
performance.

• The general principle is to give the shuffle as much memory as possible.

• There is a trade-off, in that you need to make sure that your map and
reduce functions get enough memory to operate.

• Write map and reduce functions to use as little memory as possible,

should not use an unbounded amount of memory.

• The amount of memory given to the JVMs in which the map and reduce
tasks run is set by the [Link] property.

• To make this as large as possible for the amount of memory on your task
nodes.
On the map side
• The best performance can be obtained by avoiding multiple spills to disk;
one is optimal.

• If you can estimate the size of your map outputs, then you can set the
[Link].* properties appropriately to minimize the number of spills.

• There is a MapReduce counter that counts the total number of records

that were spilled to disk over the course of a job, which can be useful for
tuning.

• The counter includes both map and reduces side spills.

On the reduce side
• The best performance is obtained when the intermediate data can reside
entirely in memory.

• By default, this does not happen, since for the general case all the memory
is reserved for the reduce function.

IV-II SEMESTER 2018-19 CSE

 Big Data 16

• If your reduce function has [Link] to 0 and a

performance boost.
light memory requirements, then setting [Link]
to 1.0 may bring

• Hadoop uses a buffer size of 4 KB by default, which is low, so you should

increase this across the cluster.
5.10 MapReduce Types
The map and reduce functions in Hadoop MapReduce have the following
general form:

map: (K1, V1) → list(K2, V2)

reduce: (K2, list(V2)) → list(K3, V3)

• The map input key and value types (K1 and V1) are different from the map
output types (K2 and V2).

• The reduce input must have the same types as the map output, although
the reduce output types may be different again (K3 and V3).
The Java API mirrors this general form:
public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {

public class Context extends MapContext<KEYIN, VALUEIN,

KEYOUT, VALUEOUT> { // ...

protected void map(KEYIN key, VALUEIN value, Context

context) throws IOException, InterruptedException {

// ...

IV-II SEMESTER 2018-19 CSE

 Big Data 17

public class Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {

public class Context extends ReducerContext<KEYIN,

VALUEIN, KEYOUT, VALUEOUT> { // ...

protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context

context) throws

IOException, InterruptedException {

// ...
}
}

• The context objects are used for emitting key-value pairs, so they are
parameterized by the output types, so that the signature of the write()
method is:

public void write(KEYOUT key, VALUEOUT value) throws IOException,

InterruptedException
• Mapper and Reducer are separate classes the type parameters have different
scopes, and the actual type argument of KEYIN (say) in the Mapper may be
different to the type of the type parameter of the same name (KEYIN) in the
Reducer.
Ex: in the maximum temparature example from earlier chapters, KEYIN is
replaced by LongWritable for the Mapper, and by Text for the Reducer.

IV-II SEMESTER 2018-19 CSE

 Big Data 18

• The map output types and the reduce input types must match
If combiner function is used, then it is the same form as the reduce function
(and is an implementation of Reducer), except its output types are the
intermediate key and value types (K2 and V2), so they can feed the reduce
function:

map: (K1, V1) → list(K2, V2)

combine: (K2, list(V2)) → list(K2, V2)

reduce: (K2, list(V2)) → list(K3, V3)

• combine and reduce functions are the same, in which case, K3 is the same
as K2, and V3 is the same as V2.

• The partition function operates on the intermediate key and value types (K2
and V2), and returns the partition index.

• The partition is determined by the key (the value is ignored):

partition: (K2, V2) → integer

In Java:

public abstract class Partitioner<KEY, VALUE> {

public abstract int getPartition(KEY key, VALUE value, int
numPartitions);

}
5.11 Input Formats
Hadoop can process many different types of data formats, from flat text files
to databases.

IV-II SEMESTER 2018-19 CSE

 Big Data 19

1) Input Splits and Records:

• An input split is a chunk of the input that is processed by a single map.
Each map processes a single split.

• Each split is divided into records, and the map processes each record—a
key-value pair—in turn.

public abstract class InputSplit {

public abstract long getLength() throws IOException,

InterruptedException; public abstract String[]
getLocations() throws IOException, InterruptedException;
}
FileInputFormat:
• FileInputFormat is the base class for all implementations of InputFormat
that use files as their data source.

• It provides two things: a place to define which files are included as the input
to a job, and an implementation for generating splits for the input files.
FileInputFormat input paths:
• The input to a job is specified as a collection of paths, which offers great
flexibility in constraining the input to a job.

• FileInputFormat offers four static convenience methods for setting a Job’s

input paths:

public static void addInputPath(Job job, Path path)

public static void addInputPaths(Job job, String

commaSeparatedPaths)

public static void setInputPaths(Job job, Path... inputPaths)

IV-II SEMESTER 2018-19 CSE

 Big Data 20

public static void setInputPaths(Job job, String

commaSeparatedPaths)
• The addInputPath() and addInputPaths() methods add a path or paths to the
list of inputs.

• The setInputPaths() methods set the entire list of paths in one go

• To exclude certain files from the input, you can set a filter using the
setInputPathFilter() method
public static void setInputPathFilter(Job job, Class<? extends
PathFilter> filter)

Fig: InputFormat class hierarchy

IV-II SEMESTER 2018-19 CSE

 Big Data 21

Paths and filters can be set through configuration properties

Table: Input path and filter properties

FileInputFormat input splits:

• FileInputFormat splits only large files. Here “large” means larger than an
HDFS block. The split size is normally the size of an HDFS block.

Table: Properties for controlling split size

Preventing splitting:
• There are a couple of ways to ensure that an existing file is not split.
First way is to increase the minimum split size to be larger than the largest
file in your system. Second way is to subclass the concrete subclass of
FileInputFormat that you want to use, to override the isSplitable() method to
return false.

IV-II SEMESTER 2018-19 CSE

 Big Data 22

File information in the mapper:

• A mapper processing a file input split can find information about the split
by calling the getInputSplit() method on the Mapper’s Context object.

Table: File split properties

Processing a whole file as a record:

• A related requirement that sometimes crops up is for mappers to have
access to the full contents of a file. The listing for WholeFileInputFormat
shows a way of doing this.

Ex : An InputFormat for reading a whole file as a record

public class WholeFileInputFormat extends

FileInputFormat<NullWritable, BytesWritable> { @Override
protected boolean isSplitable(JobContext
context, Path file) { return false;
}
}
• WholeFileRecordReader is responsible for taking a FileSplit and converting it
into a single record, with a null key and a value containing the bytes of the
file.
2) Text Input
• Hadoop can process unstructured text. It provide different InputFormat to
process text.

IV-II SEMESTER 2018-19 CSE

 Big Data 23

TextInputFormat:
• TextInputFormat is the default InputFormat. Each record is a line of input.

• The key, a LongWritable, is the byte offset within the file of the beginning of
the line.

• The value is the contents of the line, excluding any line terminators
(newline, carriage return), and is packaged as a Text object.

• A file containing the following text:

On the top of the Crumpetty Tree
The Quangle Wangle sat,
But his face you could not see,
On account of his Beaver Hat.
is divided into one split of four records. The records are interpreted as the
following key-value pairs:
(0, On the top of the Crumpetty Tree)
(33, The Quangle Wangle sat,)
(57, But his face you could not see,)
(89, On account of his Beaver Hat.)
Fig: Logical records and HDFS blocks for TextInputFormat

IV-II SEMESTER 2018-19 CSE

 Big Data 24

KeyValueTextInputFormat:
• TextInputFormat’s keys, being simply the offset within the file, are not
normally very useful.

• It is common for each line in a file to be a key-value pair, separated by a

delimiter such as a tab character by default.

• Specify the separator via the

[Link] property.

• Consider the following input file, where → represents a (horizontal) tab

character:

line1→On the top of the Crumpetty Tree

line2→The Quangle Wangle sat,

line3→But his face you could not see,

line4→On account of his Beaver Hat.

• Like in the TextInputFormat case, the input is in a single split comprising

four records, although this time the keys are the Text sequences before the
tab in each line:
(line1, On the top of the
Crumpetty Tree) (line2, The
Quangle Wangle sat,)
(line3, But his face you
could not see,) (line4, On
account of his Beaver Hat.)

IV-II SEMESTER 2018-19 CSE

 Big Data 25

NLineInputFormat:
• N refers to the number of lines of input that each mapper receives.

• With N set to one, each mapper receives exactly one line of input.
[Link] property controls the value of
N.

Ex: N is two, then each split contains two lines. One mapper will receive the
first two key-value pairs:

(0, On the top of the

Crumpetty Tree) (33, The
Quangle Wangle sat,)

And another mapper will receive the second two key-value pairs:

(57, But his face you

could not see,) (89, On
account of his Beaver
Hat.)
3) Binary Input:
• Hadoop MapReduce is not just restricted to processing textual data—it has
support for binary formats, too.

• SequenceFileInputFormat: Hadoop’s sequence file format stores sequences

of binary key-value pairs.

• SequenceFileAsTextInputFormat: SequenceFileAsTextInputFormat is
a variant of

IV-II SEMESTER 2018-19 CSE

 Big Data 26

SequenceFileInputFormat that converts the sequence file’s keys and values

to Text objects.

• SequenceFileAsBinaryInputFormat: SequenceFileAsBinaryInputFormat
is a variant of

SequenceFileInputFormat that retrieves the sequence file’s keys and values

as opaque binary objects.
4) Multiple Inputs:
• Although the input to a MapReduce job may consist of multiple input files,
all of the input is interpreted by a single InputFormat and a single Mapper.

• The MultipleInputs class has an overloaded version of addInputPath() that

doesn’t take a mapper: public static void addInputPath(Job job, Path
path, Class<? extends InputFormat> inputFormatClass)
5.12 Output Formats
• Hadoop has output data formats that correspond to the input formats

Figure: OutputFormat class hierarchy

IV-II SEMESTER 2018-19 CSE

 Big Data 27

1) Text Output:
• The default output format, TextOutputFormat, writes records as lines of
text.

• Its keys and values may be of any type, since TextOutputFormat turns
them to strings by calling toString() on them.

• Each key-value pair is separated by a tab character, that may be changed

using the [Link] property.
2) Binary Output
• SequenceFileOutputFormat: As the name indicates,
SequenceFileOutputFormat writes sequence files for its output.
Compression is controlled via the static methods on
SequenceFileOutputFormat.

• SequenceFileAsBinaryOutputFormat:
SequenceFileAsBinaryOutputFormat is the counterpart to
SequenceFileAsBinaryInput Format, and it writes keys and values in raw
binary format into a SequenceFile container.

• MapFileOutputFormat: MapFileOutputFormat writes MapFiles as output.

The keys in a MapFile must be added in order, so you need to ensure that
your reducers emit keys in sorted order.

3) Multiple Outputs:

• FileOutputFormat and its subclasses generate a set of files in the output

directory.

IV-II SEMESTER 2018-19 CSE

 Big Data 28

• There is one file per reducer, and files are named by the partition number:
part-r-00000, partr-00001, etc.

• MapReduce comes with the MultipleOutputs class to do this.

Zero reducers: There are no partitions, as the application needs to run only
map tasks.

One reducer: It can be convenient to run small jobs to combine the output
of previous jobs into a single file when the amount of data is small enough
to be processed.

MultipleOutputs:

• MultipleOutputs allows you to write data to files whose names are derived
from the output keys and values.

• This allows each reducer (or mapper in a map-only job) to create more than
a single file.

• File names are of the form name-m-nnnnn for map outputs and name-r-
nnnnn for reduce outputs

• name is an arbitrary name that is set by the program, and nnnnn is an

integer designating the part number, starting from zero.

• The part number ensures that outputs written from different partitions
Lazy Output:
• FileOutputFormat subclasses will create output (part-r-nnnnn) files, even if
they are empty.

IV-II SEMESTER 2018-19 CSE

 Big Data 29

• Some applications prefer that empty files not be created, which is where
LazyOutputFormat helps.

• It is a wrapper output format that ensures that the output file is Output
Formats created only when the first record is emitted for a given partition.

Database Output:

• The output formats for writing to relational databases and to HBase.

IV-II SEMESTER 2018-19 CSE

 Big Data 30

UNIT-V
Assignment-Cum-Tutorial Questions
SECTION-A
Objective Questions
1. Number of mappers is decided by the_____________. [ ]
A) Mapper specified by the programmer C) Input Splits
B) Available Mapper slots D) Input Format.
2. Map Reduce job can be written in_________. [ ]
A) Java C) Pyton
B) Ruby D) Any Language which can read from input stream
3. YARN also called as_________________. [ ]
A)MapReduce1 B) MapReduce2 C) MapReduce3 D) None
4. Expansion of YARN___ Yet Another Resource Navigator [T/F]
5. Classic MapReduce frame work also called as______ [ ]
A)MapReduce1 B) MapReduce2 C) MapReduce3 D) None
6. Input to every reducer is sorted by________ [ ]
A) Value B) key C) Both key-value pair D) key or value
7. The process of that performs sort and transfers the map outputs to the
reducers as input is ____________. [ ]
A) sort B) copy C) shuffle D) transfer
8. The default buffer size________. [ ]
A) 100MB B)64MB C)512MB D)128MB
9. Processing a whole file as a record by using isSplitable method that return
false. [T/F]
10. Number of copier threads can be changed by setting_____property.
[Link] amount of memory given to the JVM in which the map and reduce tasks
run is set by the ___________ property. [ ]
A) [Link] C) [Link]
B) [Link] D) None
12. Spills are written in ___________ fashion. [ ]
A) Sequence B) Round-Robin C) FCFS D) None

IV-II SEMESTER 2018-19 CSE

 Big Data 31

13. The default input format is [ ]

A) binary input format B) file input format
C) Text input format D) None
14. ____ is the base class for all implementations of inputFormat tha use files as
their data source. [ ]
A) BinaryInputFormat C) TextInputFormat
B) FileInputFormat D) None
15. Which static convenience method used for setting a job’s input paths.
A) addInputPaths() C) setInputPaths() [ ]
B) addInputPath() D) All
16. Default key value separator in keyValueTextInputFormat is___ [ ]
A) Tab B) White Space C) New line Character D) None
17. InNLineInputFormat N Refers to the [ ]
A) number of lined of output that each mapper returns.
B) number of lined of input that each mapper returns.
C) number of lined of output that each Reducer returns.
D) number of lined of input that each Reducer returns.
18. MultipleInputs class has an overload version of____________ that doesn’t take a
mapper. [ ]
A) setInputPath() B) getStart() C) addInputPath() D) None
19. Default output format___________ [ ]
A) BinaryoutputFormat C) TextInputFormat
B) BinaryOutputFormat D) LazyoutputFormat
20. ______ format is used for writing relational databases and HBase[ ]
A) DatabaseInput C) HBaseInput
B) DatabaseOuput D) HBaseOutput

IV-II SEMESTER 2018-19 CSE

 Big Data 32

SECTION-B
SUBJECTIVE QUESTIONS
1. Define shuffle and sort? Why it is required?
2. Write the general form of map and reduce functions and also writ the JAVA API
mirrors this general form.
3. Explain Map side Tuning properties in configuration tuning.
4. What constitutes progress in MapReduce? Explain
5. Illustrate reduce side tasks in shuffle and sort
6. What is input split? How to represent input splits? Which methods are used to
get location and length of input splits.
7. Explain how to control split size with example.
8. Elaborate classic frame work to run a Mapreduce job in Hadoop
9. Draw and explain inputformat class hierarch
10. Differentiate TextInputFormat and KeyValueTextInputFormat
11. List the Reduce-side tuning properties in configuration tuning.
12. How status updates are propagated through the MapReduce 1 system
13. Draw and explain outputformat class hierarchy
14. List the Binary output formats supported by hadoop.
15. Examine the relationship of streaming and pipes executable to the tasktracker
and its child.

IV-II SEMESTER 2018-19 CSE

 BigData 1

HIVE

HIVE

Hive is a data warehouse infrastructure tool to process structured data in

Hadoop. It resides on top of Hadoop to summarize big data, and makes
querying and analyzing easy.

Initially hive was developed by Jeff Hammerbacher at facebook, later the

apache software foundation took it up and developed it further as an open
source under the apache hive.

Hive is not

A relational database


A design for online transaction processing(OLTP)


A language for real time queries and row level updates

Features of hive

It stores schema in a database and processed data into HDFS


It is designed for OLAP


It provides SQL type language for querying called HiveQL or HQL


It is familiar, fast, scalable, and extensible

[Link]-II-Semester 2018-19 CSE

 BigData 2

5.1 Hive Shell

The shell is the primary way that we will interact with Hive, by issuing
commands in HiveQL. HiveQL is Hive’s query language, a dialect of SQL. It is
heavily influenced by MySQL, so if you are familiar with MySQL you should feel
at home using Hive.

When starting Hive for the first time, we can check that it is working by
listing its tables: there should be none. The command must be terminated with
a semicolon to tell Hive to execute it: Ex: hive> SHOW TABLES;
OK
Time taken: 10.425 seconds
Features of shell
It is possible to run the hive shell in non-interactive mode. The –f option runs
the commands in the specified file, ex: script.q,

% hive -f script.q

For short scripts, you can use the -e option to specify the commands inline, in
which case the final semicolon is not required:
% hive -e 'SELECT * FROM dummy'

Hive history file=/tmp/tom/hive_job_log_tom_201005042112_1906486281.txt

OK

Time taken: 4.734 seconds

[Link]-II-Semester 2018-19 CSE

 BigData 3

In both interactive and non-interactive mode, Hive will print information to

standard error—such as the time taken to run a query, to surpress these
messages using the –s option it shows only the result of the query.
% hive -S -e 'SELECT * FROM dummy'

Other useful Hive shell features include the ability to run commands on the
host operating system by using a! Prefix to the command and the ability to
access Hadoop filesystems using the dfs command
5.2 Hive Services
The Hive shell is only one of several services that you can run using the hive
command.
You can specify the service to run using the --service option. Type hive –service
help to get a list of available service names; the most useful are described
below.
1. cli
The command line interface to Hive (the shell). This is the default service.
2. hiveserver
Runs Hive as a server exposing a Thrift service, enabling access from a range of
clients written in different languages. Applications using the Thrift, JDBC, and
ODBC connectors need to run a Hive server to communicate with Hive. Set the
HIVE_PORT environment variable to specify the port the server will listen on
(defaults to 10,000).
3. hwi
The Hive Web Interface.
4. Jar
The Hive equivalent to hadoop jar, a convenient way to run Java
applications that includes both Hadoop and Hive classes on the
classpath.

[Link]-II-Semester 2018-19 CSE

 BigData 4

5. metastore
Using this service, it is possible to run the metastore as a standalone (remote)
process.

5.3 Hive Clients

Hive as a server (hive --service hiveserver), then there are a number of

different mechanisms for connecting to it from applications.

The relationship between Hive clients and Hive services is illustrated in

Figure

Fig: Hive architecture

Thrift Client

The Hive Thrift Client makes it easy to run Hive commands from a wide range
of programming languages. Thrift bindings for Hive are available for C++, Java,
PHP, Python, and Ruby.
JDBC Driver
Hive provides a Type 4 (pure Java) JDBC driver, defined in the class
[Link]. When configured with a JDBC URI of

[Link]-II-Semester 2018-19 CSE

 BigData 5

the form jdbc:hive://host:port/dbname, a Java application will connect to a

Hive server running in a separate process at the given host and port
ODBC Driver
The Hive ODBC Driver allows applications that support the ODBC protocol to
connect to Hive.
5.4 The metastore

The metastore is the central repository of Hive metadata.


The metastore is divided into two pieces: a service and the backing
store for the data. There are 3 different metastore configurations

1. Embedded metastore

By default, the metastore service runs in the same JVM as the Hive
service and contains an embedded Derby database instance backed by
the local disk. This is called the embedded metastore configuration


a simple way to get started with Hive


only one embedded Derby database can access the database files on disk
at any one time, which means you can only have one Hive session


if we open another session it attempts to open a connection to the
metastore i.e., trying to start a second session gives the error

Error: Failed to start database 'metastore_db'

[Link]-II-Semester 2018-19 CSE

 BigData 6

Figure : Metastore configurations

[Link] metastore

To support multiple sessions (and therefore multiple users) is to
use a standalone database.

This configuration is referred to as a local metastore, since the
metastore service still runs in the same process as the Hive
service, but connects to a database running in a separate process,
either on the same machine or on a remote machine.
3. Remote metastore
o
Where one or more metastore servers run in separate processes to
the Hive service.

o
This brings better manageability and security, since the database
tier can be completely firewalled off, and the clients no longer need
the database credentials.

[Link]-II-Semester 2018-19 CSE

 BigData 7

5.5 Comparison with traditional databases

Schema on Read versus Schema on Write

In a traditional database, a table’s schema is enforced at data load
time.

Hive does not verify the data when it is loaded but it is verified
when the query is issued

Traditional db takes longer time to load data

Schema on read makes for a very fast
initial load Query time performance:

Schema on write makes query time performance faster


Schema on read makes longer time for query execution
Transactions:
Hive doesnot support transactions.

[Link]-II-Semester 2018-19 CSE

 BigData 8

Hive does not support updates(or deletes).

Indexes:
Release 0.7.0 introduced indexes, which can speed up queries
Locking:
Release 0.7.0 introduces table and partitional level locking in hive.
Locks are managed transparently by zookeeper.
5.6 Hive QL
HiveQL is Hive’s SQL dialect.
It does not provide the full features of SQL-92 language constructs.
The main difference between HiveQL and
SQL are Table : A high-level comparison of
SQL and HiveQL

[Link]-II-Semester 2018-19 CSE

 BigData 9

Data Types

• Hive supports both primitive and complex data types.

• Primitives include

1) numeric

2) boolean

3) string,

4) timestamp

• complex data types include

1) arrays

2) maps

3) structures

[Link]-II-Semester 2018-19 CSE

 BigData 10

Operators and Functions

• Operators
1) arithmetic
2) relational
3) logical
• Categories of Functions
1) mathematical
2) statistical
3) string
4) date
5) conditional
6) aggregate
7) Functions working with XML and JSON
5.7 Tables
A Hive table is logically made up of the data being stored and the associated
metadata describing the layout of the data in the table.
The data typically resides in HDFS, although it may reside in any Hadoop
filesystem, including the local filesystem or S3.
Hive stores the metadata in a relational database—and
not in HDFS, Managed Tables and External Tables
When you create a table in Hive, by default Hive will manage the data, which
means that Hive moves the data into its warehouse directory.

[Link]-II-Semester 2018-19 CSE

 BigData 11

For example: CREATE TABLE managed_table (dummy STRING);

LOAD DATA INPATH '/user/tom/[Link]' INTO table managed_table; will move

the file hdfs://user/tom/[Link] into Hive’s warehouse directory for the
managed_table table, which is hdfs://user/hive/warehouse/managed_table

If the table is later dropped, using:

DROP TABLE managed_table;

then the table, including its metadata and its data, is deleted.

An external table behaves differently. You control the creation and deletion of
the data.

The location of the external data is specified at table creation time:

CREATE EXTERNAL TABLE external_table (dummy STRING)

LOCATION '/user/tom/external_table';

LOAD DATA INPATH '/user/tom/[Link]' INTO TABLE external_table;

Partitions and Buckets

Hive organizes tables into partitions, a way of dividing a table into coarse-
grained parts based on the value of a partition column, such as date. Tables or
partitions may further be subdivided into buckets, to give extra structure to the
data that may be used for more efficient queries.
Partitions
Hive organizes tables in to partitions.
[Link]-II-Semester 2018-19 CSE
 BigData 12

A way of dividing a table in to related parts based on the value of a

partition column ex: date, city, dept

Faster to do queries on slices of the data

Tables or partitions are sub-divided into buckets , to provide extra

structure to the data that may be used for more efficient querying

Bucketing works based on the value of hash function of some

column of a table. Ex:

A table named Tab1 contains employee data such as id, name, dept, and
yoj , need to retrieve the details of all employees who joined in 2012.

• A query searches the whole table for the required information.

• if you partition the employee data with the year and store it in a separate
file, it reduces the query processing time.

The following file contains employee data table.

/tab1/employeedata/file1 id, name, dept, yoj

1, gopal, TP,2012
2, kiran,HR,2012
3, kaleel,SC,2013
4,Prasanth,SC,20

[Link]-II-Semester 2018-19 CSE

 BigData 13

The above data is partitioned into two files using year.

/tab1/employeedata/2012/file2/tab1/employeedata/2013/file3
1, gopal, TP, 2012 3, kaleel,SC, 2013

2, kiran, HR, 2012 4, Prasanth, SC, 2013

Adding a Partition :-
We can add partitions to a table by altering the
table hive> ALTER TABLE employee

> ADD PARTITION (year=’2013’)

> location '/2012/part2012';

Renaming a Partition

hive> ALTER TABLE employee PARTITION (year=’1203’)

> RENAME TO PARTITION (Yoj=’1203’);

Show Partition

hive> show partitions employee;

Dropping a Partition

hive> ALTER TABLE employee DROP [IF EXISTS]

> PARTITION (year=’1203’);

Buckets

It is a mechanism to query and examine random samples of data
[Link]-II-Semester 2018-19 CSE
 BigData 14


Break data into a set of buckets based on a hash function of a ―bucket
column


Capability to execute queries on a sub-set of random data


Doesn’t automatically enforce bucketing


User is required to specify the number of buckets by
setting # of reducer Create and use table with Buckets

hive>create table post-count(user String,

count Int) >clustered by (user) into 5
buckets;

Use the clustered by clause to specify columns to bucket on and the

number of buckets.
Storage Formats


There are two dimensions that govern table storage in Hive: the row
format and the file format. The row format dictates how rows, and the
fields in a particular row, are stored


The file format dictates the container format for fields in a row. The
simplest format is a plain text file, but there are row-oriented and
column-oriented binary formats available,too.


The default storage format: Delimited text


When you create a table with no ROW FORMAT or STORED AS clauses,
the default format is delimited text, with a row per line.

[Link]-II-Semester 2018-19 CSE

 BigData 15


The default row delimiter is not a tab character, but the Control-A
character from the set of ASCII control codes


The choice of Control-A, sometimes written as ^A in documentation,
came about since it is less likely to be a part of the field text than a tab
character.

The default collection item delimiter is a Control-B character, used to

delimit items in an ARRAY or STRUCT, or key-value pairs in a MAP. The
default map key delimiter is a Control-C character, used to delimit the key
and value in a MAP. Rows in a table are delimited by a newline character.

Importing Data
o INSERT OVERWRITE TABLE

a table with data from another Hive table using an INSERT
statement i.e

Example

hive> INSERT OVERWRITE TABLE

target >SELECT col1, col2

>FROM
source;
Another way

Hive> FROM source

>INSERT OVERWRITE TABLE target

>SELECT col1, col2;

• Multitable insert

it’s possible to have multiple INSERT clauses in the same query.

[Link]-II-Semester 2018-19 CSE

 BigData 16


multitable insert is more efficient than multiple INSERT statements,


the source table need only be scanned once to produce the multiple,
disjoint outputs hive> FROM records2

>INSERT OVERWRITE TABLE

stations_by_year
>SELECT year, COUNT(DISTINCT station)
>GROUP BY year

>INSERT OVERWRITE TABLE

records_by_year
>SELECT year, COUNT(1)
>GROUP BY year
>INSERT OVERWRITE TABLE
good_records_by_year
>SELECT year, COUNT(1)
>WHERE temperature != 9999
>AND (quality = 0 OR quality = 1 OR quality = 4 OR quality = 5 OR
quality = 9)
>GROUP BY year;


There is a single source table (records2), but three tables to hold the
results from three different queries over the source.

o CREATE TABLE...AS SELECT


To store the output of a Hive query in a new table.


The new table’s column definitions are derived from the columns
retrieved by the

[Link]-II-Semester 2018-19 CSE

 BigData 17

SELECT
clause
Example

CREATE TABLE target

AS

SELECT col1, col2

FROM source;


Alter TableAlter the attributes of a table such as changing its table
name, changing column names, adding columns, and deleting or
replacing columns.
Syntax

ALTER TABLE name RENAME TO new_name

ALTER TABLE name ADD COLUMNS (col_spec[, col_spec ...])

ALTER TABLE name DROP [COLUMN] column_name

ALTER TABLE name CHANGE column_name new_name

new_type ALTER TABLE name REPLACE COLUMNS
(col_spec[, col_spec ...]) Examples

hive> ALTER TABLE employee RENAME TO emp;

hive>ALTER TABLE employee ADD COLUMNS (dept
string); hive>ALTER TABLE employee DROP dept;

hive> ALTER TABLE employee CHANGE name ename

String; hive> ALTER TABLE employee CHANGE salary
salary Double;
[Link]-II-Semester 2018-19 CSE
 BigData 18

hive> ALTER TABLE employee REPLACE COLUMNS ( eid INT empid Int,
ename STRING name String);

hive> DROP TABLE IF EXISTS employee;

Select Statements

SELECT statement is used to retrieve the data from a table. WHERE

clause works similar to a condition. It filters the data using the condition
and gives you a finite result. Syntax:

SELECT [ALL | DISTINCT] select_expr, select_expr, ...

FROM table_reference

[WHERE where_condition]

[GROUP BY col_list]

[HAVING having_condition]

[CLUSTER BY col_list |

[DISTRIBUTE BY col_list]

[SORT BY col_list]]

[LIMIT number];

[Link]-II-Semester 2018-19 CSE

 BigData 19

Examples:

hive> SELECT * FROM employee WHERE

eid=1205; hive> SELECT * FROM employee
WHERE
salary>=40000;

hive> SELECT Id, Name, Dept FROM employee ORDER

BY DEPT;
hive> SELECT Dept,count(*) FROM employee GROUP BY
DEPT;
hive> FROM records2
SELECT year, temperature

DISTRIBUTE BY year
SORT BY year ASC, temperature
DESC;
hive> SELECT * FROM employee LIMIT 4;
5.8 querying data
Sorting and Aggregating
Sorting data in Hive can be achieved by use of a standard ORDER BY clause,
but there is a catch. ORDER BY produces a result that is totally sorted, as
expected, but to do so it sets the number of reducers to one, making it very
inefficient for large datasets.
In some cases, you want to control which reducer a particular row goes to,
typically so you can perform some subsequent aggregation. This is what Hive’s
DISTRIBUTE BY clause does. Here’s an example to sort the weather dataset by
year and temperature
SORT BY produces a sorted file per reducer.
hive> FROM records2
> SELECT year, temperature
> DISTRIBUTE BY year

[Link]-II-Semester 2018-19 CSE

 BigData 20

> SORT BY year ASC, temperature

DESC; 1949 111 1949 78 1950 22
1950 0 1950 -11
MapReduce Scripts
Using an approach like Hadoop Streaming, the TRANSFORM, MAP, and
REDUCE clauses make it possible to invoke an external script or program from
Hive.
Example: Python script to filter out poor quality weather records
#!/usr/bin/env python
import re

import sys

for line in [Link]:

(year, temp, q) = [Link]().split()

if (temp != "9999" and [Link]("[01459]", q)):

print "%s\t%s" % (year, temp)

We can use the script as follows:

hive> ADD FILE /path/to/is_good_quality.py;

hive> FROM records2

> SELECT TRANSFORM(year, temperature, quality)

> USING 'is_good_quality.py'

[Link]-II-Semester 2018-19 CSE

 BigData 21

> AS year,
temperature; 1949
111 1949 78 1950 0

1950 22

1950 -11
Before running the query, we need to register the script with Hive. This is so
Hive knows to ship the file to the Hadoop cluster

The query itself streams the year, temperature, and quality fields as a tab-
separated line to the is_good_quality.py script, and parses the tab-separated
output into year and temperature fields to form the output of the query.

This example has no reducers. If we use a nested form for the query, we can
specify a map and a reduce function. This time we use the MAP and REDUCE
keywords, but SELECT TRANSFORM in both cases would have the same
result. The source for the max_temperature_reduce.py script is shown in
Example
FROM
(FROMrecos
2

MAP year, temperature,

quality USING
'is_good_quality.py'

AS year, temperature)
map_output REDUCE year,
temperature USING
'max_temperature_reduce.py'
AS year, temperature;

[Link]-II-Semester 2018-19 CSE

 BigData 22

Views

A view is a sort of ―virtual table that is defined by a SELECT statement.


Views can be used to present data to users in a different way to the way
it is actually stored on disk.


Views may also be used to restrict users access to particular subsets of
tables that they are authorized to see.


First create table and then insert data into it

hive> create table posts(id int,name string,sal double)

> row format delimited

> fields terminated by ','

stored as textfile;


Create View

hive> create view posts_name as

> select name from posts;

hive> create view first_id as

select * from posts whereid=1;

hive> create view max_sal as

select name ,max(sal) from posts;

Show views

[Link]-II-Semester 2018-19 CSE

 BigData 23

hive> show tables;

Altering views

hive> alter view first_id rename to 1stid;

Drop a view
hive> drop view
1stid;
Joins


JOIN is a clause that is used for combining specific fields from two tables
by using values common to each one.


used to combine records from two or more tables in the database.


similar to SQL JOINS.


There are different types of joins given as follows:

o
JOIN

o
LEFT OUTER JOIN

o
RIGHT OUTER JOIN

o
FULL OUTER JOIN


Can join multiple tables


Default join Is Inner join

Rows are joined where the keys match

[Link]-II-Semester 2018-19 CSE

 BigData 24

Rows that do not have matches are not included in the result

The simplest kind of join is the inner join, where each match in the input
tables results in a row in the output table it is being joined to (things): hive>
SELECT * FROM sales;

joe 2
Hank 4
Ali 0
Eve 3
Hank 2
hive> SELECT * FROM things;
2 Tie
4 Coat
3 Hat
1 Scarf
hive> SELECT sales.*, things.*

> FROM sales JOIN things ON ([Link] = [Link]);

Joe 2 2 Tie

Hank 2 2 Tie

Eve 3 3 Hat

Hank 4 4 Coat


The table in the FROM clause (sales) is joined with the table in the JOIN
clause (things), using the predicate in the ON clause

[Link]-II-Semester 2018-19 CSE

 BigData 25


Hive only supports equijoins, which means that only equality can be
used in the join predicate, which here matches on the id column in both
tables.

 the row for Ali did not appear in the output, since the ID of the item she
purchased was not present in the things table
Left Outer Join

Outer joins allow you to find non matches in the tables being joined.

If we change

the join type to LEFT OUTER JOIN, then the query will return a row for
every row in the left table (sales), even if there is no corresponding row in
the table it is being joined to (things):

hive> SELECT * FROM sales;

Joe 2
Hank 4
Ali 0
Eve 3
Hank 2

hive> SELECT * FROM things;

2 Tie

4 Coat

3 Hat
1 Scarf

[Link]-II-Semester 2018-19 CSE

 BigData 26

The row for Ali is now returned, and the columns from the things table are
NULL, since there is no match.

i.e Row from the first table are included whether they have a match or not.
Columns from the unmatched(second) table are set to null.
Right Outer Join


Opposite of Left Outer Join, Rows from the second table are included no
matter what. Columns from the unmatched (first) table are set to null.


all items from the things table are included, even those that weren’t
purchased by anyone
(a scarf):
hive> SELECT * FROM sales;

Joe 2
Hank 4
Ali 0
Eve 3
Hank 2

hive> SELECT * FROM things;

2 Tie

4 Coat

3 Hat
1 Scarf
We can perform left outer join on the two tables as follows:
hive> SELECT sales.*, things.*

[Link]-II-Semester 2018-19 CSE

 BigData 27

> FROM sales RIGHT OUTER JOIN things ON ([Link] =

[Link]); NULL NULL 1 Scarf

Joe 2 2 Tie
Han 2 2
k Tie
Eve 3 3 Hat
Coa
Hak 4 4 t
Full Outer Join

Rows from both sides are includes. For unmatched rows the columns
from the other table are set to null

In full outer join, the output has a row for each row from both tables in
the join:

hive> SELECT * FROM sales;

Joe 2
Hank 4
Ali 0
Eve 3
We can perform left outer join on the two tables as follows:
hive> SELECT sales.*, things.*

[Link]-II-Semester 2018-19 CSE

 BigData 28

Ali

Joe

HanEve

Hank

[Link]-II-Semester 2018-19 CSE

 BigData 29

> FROM sales LEFT OUTER JOIN

0 NULL NULL

2 2 Tie

2 2 Tie

3 3 Hat

4 4 Coat
Subqueries

A subquery is a SELECT statement that is embedded in another SQL
statement.


Hive has limited support for subqueries

The query finds the mean maximum temperature for every year and weather
station:

SELECT station, year, AVG(max_temperature)

FROM (

SELECT station, year, MAX(temperature) AS

max_temperature FROM records2

WHERE temperature != 9999

AND (quality = 0 OR quality = 1 OR quality = 4 OR quality = 5 OR quality =

9)

[Link]-II-Semester 2018-19 CSE

 BigData 30

GROUP BY station, year

) mt
GROUP BY station, year;


The subquery is used to find the maximum temperature for each
station/date combination,


the outer query uses the AVG aggregate function to find the average of
the maximum temperature readings for each station/date combination.


The outer query accesses the results of the subquery like it does a table,
which is why the subquery must be given an alias (mt).


The columns of the subquery have to be given unique names so that the
outer query can refer to them.
5.9 User-Defined functions

Write the query that can’t be expressed easily using built-in functions.


Write a User-Defined Function(UDF) .


Easy to plug in own processing code and invoke it from a Hive Query.


There are 3 types of UDF in Hive
1) Regular UDFs
Operates on a single row and produces a single row as its output. Ex:
Mathematical and String functions

2) UDAF (User-defined aggregate functions)

works on multiple input rows and creates a single output row.

Aggregate functions include such functions as COUNT and MAX.

[Link]-II-Semester 2018-19 CSE

 BigData 31

3) UDTFs (user-defined table-generating functions)

operates on a single row and produces multiple rows—a table—as
output
5.9 User-Defined functions

Write the query that can’t be expressed easily using built-in functions.


Write a User-Defined Function(UDF) .


Easy to plug in own processing code and invoke it from a Hive Query.


There are 3 types of UDF in Hive
2) Regular UDFs
Operates on a single row and produces a single row as its output. Ex:
Mathematical and String functions
4) UDAF (User-defined aggregate functions)
o
works on multiple input rows and creates a single output row.

o
Aggregate functions include such functions as COUNT and MAX.

5) UDTFs (user-defined table-generating functions)

operates on a single row and produces multiple rows—a table—as
output

[Link]-II-Semester 2018-19 CSE

 BigData 32

UNIT-VI
Assignment-Cum-Tutorial Questions
SECTION-A
Objective Questions
1. Which of the following command sets the value of a particular configuration
variable [ ]
A) Set-v B) set <key>=<value> C) set D) reset
2. Which of the following operator executes a shell command from the Hive
shell? [ ]
A) | B) ! C^ D) +
3. Which of the following will remove the resource(s) from the distributed
cache? [ ]
A) Delete FILE[S] <filepath>*
B) Delete JAR[S]<filepath>*
C) Delete ARCHIVE[S]<filepath>*
D) All
4. ________ is a shell utility which can be used to run Hive queries in either
interactive or batch mode. [ ]
A) $HIVE/bin/hive
B) $HIVE_HOME/hive
C) $HIVE_HOME/bin/hive
D) All
5. Which of the following is a command line option? [ ]
A) –d,-define <key=value>
B) –e,-define<key=value>
C) –f,-define<key=value>
D) None
6. Hive uses___________ for logging [ ]
A)logj4 B) log41 C) log4i D) log4j
7. Hive Server2 introduced in HIVE 0.11 has new CLI called [ ]
A) BeeLine B) SQLLine C)HIVELine D) CLILine
8. Hcatalog is installed with HIVE, starting with HIVE relase [ ]
A) 0.10.0 B) 0.9.0 C)0.11.0 D)0.1.20

[Link]-II-Semester 2018-19 CSE

 BigData 33

9. _____ supports a new command shell Beeline that works with HIVE Server2.
[ ]
A) HiveServer2 B) HiveServer3 C) HiveServer4 D) None
10. In _______ mode HiveServer2 only accepts valid Thrift calls. [ ]
A) Remote B) HTTP C) Embedded D) Interactive
11. Hive specific commands can be run from Beeline, When the Hive _____
driver is used. [ ]
A) ODBC B) JDBC C) ODBC-JDBC D) ALL
12. The ___ allow users to read or write Avro data s Hive Table [ ]
A) AvroSerde B) HiveSerde C) SQLSerde D) None
13. Starting in Hive_____ the Avro schema can be inferred from the hive table
schema. [ ]
A) 0.14 B) 0.12 C) 0.13 D) 0.11
14. Which of the following data type is supported by HIVE [ ]
A) map B) record C) string D) enum
15. which of the following data type is converted to Array prior to Hive 0.12.0
[ ]
A) map B) long C) float D) bytes
[Link]-backed tables can simply be created by using _________ in a DDL
statement. [ ]
A) “STORED AS AVERO” C. –STORED AS AVROHIVE
B) –STORED AS HIVE D. –STORED AS SERED
17. Types that may be null must be defined as a ___________ of that type and
NULL within AVRO. [ ]
A) Union B) intersection C) Set D) All
18. use_____ and embed the schema in the create statement [ ]
A) [Link] B) [Link] C) [Link] D) All
19. Serialization of string columns uses a____ to form unique column values.
A) Footer B) STRIPES C) Dictionary D) Index
20. Hive uses________ -Style escaping within the strings [ ]
A) C B) JAVA C) python D) Scala

[Link]-II-Semester 2018-19 CSE

 BigData 34

SECTION-B

SUBJECTIVE QUESTIONS
1. What is hive? List the features of hive?
2. List out hive Services
3. What is metastore? What are different types of metastores?
4. What are megastore configuration properties?
5. Comapre the SQL and HIVEQL
6. List out Hive Data Types?
7. Explain about partitions and buckets?
8. Outline about Querying Data?
9. What are user-defined functions?
10. Explain joins?
11. Explain about HIVEQL in Hadoop System
12. Illustrate the HIVE Shell?
13. Describe about the tables in HIVE.
14. Explain about HIVE architecture?
15. Compare HIVE with traditional database?
16. Elaborate on HIVE QL data manipulation and queries in details
17. Discuss about the relationship between HIVE clients and HIVE Services
with a neat diagram?
18. Explain in detail about Map side and Reduce Side joins.

[Link]-II-Semester 2018-19 CSE