Map Reduce Workflows
• To deal with complexity
• Increase no of jobs rather increasing functions
• Hive,pig
Decomposing a Problem into MapReduce
Jobs
• To find the mean maximum recorded
temperature for every day of the year and every
weather station
• The computation decomposes most naturally
into two stages:
1. Compute the maximum daily temperature for
every station-date pair.
2. Compute the mean of the maximum daily
temperatures for every station-day- month key.
• The output from first stage looks like this for
the station we are interested in (the
mean_max_daily_temp.sh
• 029070-99999 19010101 0
• 029070-99999 19020101 -94
• ….
• The second stage averages these daily maxima
over years to yield:
• 029070-99999 0101 -68
• map and reduce functions even more
composable
• splitting these into distinct mappers and
chaining them
• single mapper using the ChainMapper
Combined with a ChainReduce
Job Control
• linear chain of jobs
• complex directed acyclic graph (DAG) of jobs
• linear chain-job one after another
[Link](conf1);
[Link](conf2);
• IOException
• For more complex jobs
• [Link]
package:
• complex workflow
• Apache Oozie-([Link]
• [Link]/oozie/)
Apache oozie
• Oozie has two main parts:
• a workflow engine - Stores and runs
workflows composed of Hadoop jobs
• coordinator engine-runs workflow jobs based
on predefined schedules and data availability
• oozie runs as a service in the cluster, and
clients submit a workflow definitions for
immediate or later execution
• workflow is a DAG of action nodes and control-
flow nodes.
• action node performs a workflow task
• A control-flow node governs the workflow
execution between actions by allowing such
constructs as conditional logic
• When the workflow completes, Oozie can
make an HTTP call- back to the client to inform
it of the workflow status
• <workflow-app xmlns="uri:oozie:workflow:0.1" name="max-temp-workflow"> <start to="max-temp-mr"/>
• <action name="max-temp-mr">
• <map-reduce>
• <job-tracker>${jobTracker}</job-tracker>
• <name-node>${nameNode}</name-node>
• <prepare>
• <delete path="${nameNode}/user/${wf:user()}/output"/> </prepare>
• <configuration>
• <property>
• <name>[Link]</name>
• <value>OldMaxTemperature$OldMaxTemperatureMapper</value> </property>
• <property>
• <name>[Link]</name>
• <value>OldMaxTemperature$OldMaxTemperatureReducer</value> </property>
• <property>
• <name>[Link]</name>
• <value>OldMaxTemperature$OldMaxTemperatureReducer</value> </property>
• <property>
• <name>[Link]</name> <value>[Link]</value> </property>
• <property>
• a start control node,
• a kill control node
• end control node
• <name>[Link]</name>
• <value>[Link]</value>
• </property>
• <property>
• <name>[Link]</name>
• <value>/user/${wf:user()}/input/ncdc/micro</value>
• </property>
• <property>
• <name>[Link]</name>
• <value>/user/${wf:user()}/output</value>
• </property>
• </configuration>
• </map-reduce>
• <ok to="end"/>
• <error to="fail"/>
• </action>
• <kill name="fail">
• <message>MapReduce failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message> </kill>
• <end name="end"/>
• </workflow-app>
Transition diagram of an Oozie workflow.
Running an Oozie workflow job
• export the OOZIE_URL environment
% export OOZIE_URL="[Link]
alhost:11000/oozie“
• To run the workflow job:
% oozie job -config
ch05/src/main/resources/max-temp-
[Link] -run
…contd
• nameNode=hdfs://localhost:8020
• jobTracker=localhost:8021
• [Link]=${nameNode}/
user/${[Link]}/max-temp-workflow
• get information about the status of the
workflow job we use the -info option
• % oozie job -info 0000009-120119174508294-
oozie-tom-W
..contd
• % hadoop fs -cat output/part-*
• 1949 111
• 1950 22
Writing a unit test with MRunit
• Mockito-Java mock object frameworks
• To verify the ouput
• Mapper
• Unit test for MaxTemperatureMapper
Example 5-4. Unit test for MaxTemperatureMapper
import static [Link].*;
import [Link];
import [Link].*;
import [Link].*;
public class MaxTemperatureMapperTest {
@Test
public void processesValidRecord() throws IOException,
InterruptedException {
MaxTemperatureMapper mapper = new MaxTemperatureMapper();
Text value = new
Text("0043011990999991950051518004+68750+023550FM-12+0382" +
// Year ^^^^
"99999V0203201N00261220001CN9999999N9-00111+99999999999");
// Temperature ^^^^^
[Link] context =
mock([Link](context).write(new
Text("1950"), new IntWritable(-11));
}
}[Link]);
[Link](null, value, context);
public void ignoresMissingTemperatureRecord() throws
IOException,
InterruptedException {
MaxTemperatureMapper mapper = new
MaxTemperatureMapper();
Text value = new
Text("0043011990999991950051518004+68750+023550FM-
12+0382" + // Year ^^^^
"99999V0203201N00261220001CN9999999N9+99991+99999999
999"); // Temperature ^^^^^
[Link] context =
mock([Link])
[Link](null, value, context);
verify(context, never()).write(any([Link]),
any([Link])); }
Reducer
• Reducer for maximum temperature example
• public class MaxTemperatureReducer
• extends Reducer<Text, IntWritable, Text, IntWritable> {
• @Override
• public void reduce(Text key, Iterable<IntWritable> values,
Context context)
• throws IOException, InterruptedException { int maxValue
= Integer.MIN_VALUE; for (IntWritable value : values)
{ maxValue = [Link](maxValue, [Link]()); }
• [Link](key, new IntWritable(maxValue)); }
• }
public class MaxTemperatureReducer
extends Reducer<Text, IntWritable, Text, IntWritable> {
@Override
public void reduce(Text key, Iterable<IntWritable> values,
Context context)
throws IOException, InterruptedException { int
maxValue = Integer.MIN_VALUE; for (IntWritable value :
values) {
maxValue = [Link](maxValue, [Link]()); }
[Link](key, new IntWritable(maxValue)); }
}
Failures in YARN
• the task,
• the application master
• the node manager and
• the resource manager.
Shuffle and Sort
• The process by which the system performs the
sort—and transfers the map outputs to the
reducers as inputs—is known as the shuffle
• Each map task has a circular memory buffer
that it writes the output to. The buffer is 100
MB by default
• [Link]
• contents of the buffer reaches a certain
threshold size
• Thread will start to spill the contents to disk.
Map outputs will continue to be written to the
buffer while the spill takes place
• Spills are written in round-robin fashion to
the directories specified by the
[Link] property
• Combiner function
• thread first divides the data into partitions
• 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.
• combiners may be run repeatedly over the
input without affecting the final result
• [Link]
Reducer side
• the reduce task needs the map output for its
particular partition from several map
• the reduce task needs the map output for its
particular partition from several map
• The map tasks may finish at different times, so
the reduce task starts copying their outputs as
soon as each completes. This is known as the
copy phase
• five threads
• [Link]
• The map outputs are copied to the reduce
task JVM’s memory otherwise they are copied
to disk.
• As the copies accumulate on disk, a
background thread merges them into larger,
sorted files
sort phase(called the merge phase)
• if there were 50 map outputs, and the merge
factor was 10 (the default, controlled by the
[Link] property
• final round that merges these five files into a
single sorted file
Efficient merging