How to convert MapReduce Job to Oozie Job

This article takes the most famous "pi" program for example, explaining the steps to convert the MapReduce job to an Oozie job using MapReduce action nodes.

hadoop jar /opt/cloudera/parcels/CDH/lib/hadoop-0.20-mapreduce/hadoop-examples.jar pi 1 1

1. Find and read source code.

Firstly we need to find the correct java source code and make sure you understand the logic.
For this example, we can find the source code from Github here.
Per ExampleDriver.java, "pi" program is called from QuasiMonteCarlo.java.
The most important function is "estimatePi", and below is the most important snippet:
    job.setJarByClass(QuasiMonteCarlo.class);
    job.setInputFormatClass(SequenceFileInputFormat.class);
    job.setOutputKeyClass(BooleanWritable.class);
    job.setOutputValueClass(LongWritable.class);
    job.setOutputFormatClass(SequenceFileOutputFormat.class);
    job.setMapperClass(QmcMapper.class);
    job.setReducerClass(QmcReducer.class);
    job.setNumReduceTasks(1);
    job.setSpeculativeExecution(false);

2. Learn from oozie-examples.tar.gz shipped with oozie binary.

For example, CDH5 has below:
/opt/cloudera/parcels/CDH/share/doc/oozie-4.0.0+cdh5.0.2+180/oozie-examples.tar.gz
Play with above examples, and learn from Oozie guide.

3. Prepare Configuration Files

3.1 Create a local directory named "map-reduce_pi" with 2 files -- job.properties and workflow.xml and 1 sub-directory -- lib.

[root@admin map-reduce_pi]# ls
job.properties  lib  workflow.xml

3.2 job.properties

job.properties normally specifies the namenode(nameservice if HA is enabled), jobtracker(or yarn resourcemanager), resource queue name, and some other parameters used by workflow.xml.
It should be on local filesystem. 
nameNode=hdfs://nameservice1
jobTracker=hdw2.xxx.com:8032
queueName=default
examplesRoot=examples
oozie.wf.application.path=${nameNode}/user/${user.name}/${examplesRoot}/apps/map-reduce_pi
outputDir=map-reduce_pi

3.3 Put needed jar file in lib sub-directory.

[root@admin lib]# ls
hadoop-mapreduce-examples-2.3.0-cdh5.0.2.jar

3.3 Identify needed class full path.

From above source code, we identified the most important classes. Now, we need to identify the full path of them and make sure all needed jar files are included.
Map and Reduce classes are in hadoop-mapreduce-examples-2.3.0-cdh5.0.2.jar in "lib" sub-directory:
[root@admin lib]# jar tvf hadoop-mapreduce-examples-2.3.0-cdh5.0.2.jar|grep -i QuasiMonteCarlo|egrep -i "reduce|map"
  4510 Mon Jun 09 09:31:30 PDT 2014 org/apache/hadoop/examples/QuasiMonteCarlo$QmcReducer.class
  3027 Mon Jun 09 09:31:30 PDT 2014 org/apache/hadoop/examples/QuasiMonteCarlo$QmcMapper.class
So the full path of above 2 classes are:
org.apache.hadoop.examples.QuasiMonteCarlo$QmcMapper
org.apache.hadoop.examples.QuasiMonteCarlo$QmcReducer
Other full path of needed classes can be found from class import part of source code:
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.LongWritable;

3.4 workflow.xml

Here I am using new api, so some properties are different than old api.
To make things simpler next time, I created below mapping between java code and property name in workflow.xml.
JAVA functionsproperty in workflow.xml
job.setMapperClassmapreduce.map.class
job.setReducerClassmapreduce.reduce.class
job.setInputFormatClassmapreduce.job.inputformat.class
job.setOutputFormatClassmapreduce.job.outputformat.class
job.setOutputKeyClassmapreduce.job.output.key.class
job.setOutputValueClassmapreduce.job.output.value.class
FileInputFormat.setInputPathsmapreduce.input.fileinputformat.inputdir
FileOutputFormat.setOutputPathmapreduce.output.fileoutputformat.outputdir
This is sample workflow.xml:
<workflow-app xmlns="uri:oozie:workflow:0.2" name="map-reduce-wf-pi">
    <start to="mr-node"/>
    <action name="mr-node">
        <map-reduce>
            <job-tracker>${jobTracker}</job-tracker>
            <name-node>${nameNode}</name-node>
            <prepare>
                <delete path="/user/root/outputforpi"/>
            </prepare>
            <configuration>
              <property>
                  <name>mapred.mapper.new-api</name>
                  <value>true</value>
              </property>
              <property>
                  <name>mapred.reducer.new-api</name>
                  <value>true</value>
              </property>
              <property>
                    <name>mapred.job.queue.name</name>
                    <value>${queueName}</value>
                </property>
                <property>
                    <name>mapreduce.map.class</name>
                    <value>org.apache.hadoop.examples.QuasiMonteCarlo$QmcMapper</value>
                </property>
                <property>
                    <name>mapreduce.reduce.class</name>
                    <value>org.apache.hadoop.examples.QuasiMonteCarlo$QmcReducer</value>
                </property>
                <property>
                    <name>mapred.map.tasks</name>
                    <value>10</value>
                </property>
                <property>
                    <name>mapred.reduce.tasks</name>
                    <value>1</value>
                </property>
                <property>
                        <name>mapred.map.tasks.speculative.execution</name>
                        <value>false</value>
                </property>
                <property>
                        <name>mapred.reduce.tasks.speculative.execution</name>
                        <value>false</value>
                </property>
                <property>
                    <name>mapreduce.job.inputformat.class</name>
                    <value>org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat</value>
                </property>
                <property>
                    <name>mapreduce.job.outputformat.class</name>
                    <value>org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat</value>
                </property>
                <property>
                    <name>mapreduce.job.output.key.class</name>
                    <value>org.apache.hadoop.io.BooleanWritable</value>
                </property>
                <property>
                    <name>mapreduce.job.output.value.class</name>
                    <value>org.apache.hadoop.io.LongWritable</value>
                </property>
                <property>
                    <name>mapreduce.input.fileinputformat.inputdir</name>
                    <value>/user/root/inputforpi</value>
                </property>
                <property>
                    <name>mapreduce.output.fileoutputformat.outputdir</name>
                    <value>/user/root/outputforpi</value>
                </property>
            </configuration>
        </map-reduce>
        <ok to="end"/>
        <error to="fail"/>
    </action>
    <kill name="fail">
        <message>Map/Reduce pi failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
    </kill>
    <end name="end"/>
</workflow-app>

4. Deploy

Put the whole directory "map-reduce_pi" to HDFS.
hdfs dfs -rmr /user/root/examples/apps/map-reduce_pi
hdfs dfs -put /root/oozie/examples/apps/map-reduce_pi /user/root/examples/apps/map-reduce_pi

5. Feed input directory

The "pi" program generates the input data in the code, so one easy way is to run the "pi" MapReduce program, and copy the input directory when it is running. 
[root@admin]# hdfs dfs -ls /user/root/inputforpi
Found 1 items
-rw-r--r--   3 root root        118 2014-07-14 11:53 /user/root/inputforpi/part0

6. Execute 

oozie job -config /root/oozie/examples/apps/map-reduce_pi/job.properties -run
Monitor the status and make sure it SUCCEEDED.
oozie job -info <job ID>

7. Troubleshoot

Please refer to Troubleshoot Oozie MapReduce jobs.

Commentaires

Enregistrer un commentaire

Posts les plus consultés de ce blog

Controlling Parallelism in Spark by controlling the input partitions by controlling the input partitions

Image
It is useful to be able to control the degree of parallelism when using Spark. Spark provides a very convenient method to increase the degree of parallelism which should be adequate in practice. This blog entry of the "Inside Spark" series describes the knobs Spark uses to control the degree of parallelism. Controlling the number of partitions in the " local " mode Spark is designed ground up to enable unit testing. To control the degree of parallelism in the local mode simply utilize the function JavaRDD<String> org.apache.spark.api.java.JavaSparkContext.parallelize(List<String> list, int numSlices) The following snippet of code demonstrates this usage- public class SparkControlPartitionSizeLocally { public static void main ( String [] args ) { JavaSparkContext sc = new JavaSparkContext ( "local" , "localpartitionsizecontrol" ); String input = "four score and sev...
Lire la suite

Spark performance optimization: shuffle tuning

Image
Article directory 1 shuffle tuning 1.1 Summary of tuning 1.2 Overview of ShuffleManager Development 1.3 HashShuffleManager operating principle 1.3.1 Unoptured HashShuffleManager 1.3.2 Optimized HashShuffleManager 1.4 SortShuffleManager operating principle 1.4.1 General operating mechanism 1.4.2 bypass running mechanism 1.5 shuffle related parameters tuning 1.5.1 spark.shuffle.file.buffer 1.5.2 spark.reducer.maxSizeInFlight 1.5.3 spark.shuffle.io.maxRetries 1.5.4 spark.shuffle.io.retryWait 1.5.5 spark.shuffle.memoryFraction 1.5.6 spark.shuffle.manager 1.5.7 spark.shuffle.sort.bypassMergeThreshold 1.5.8 spark.shuffle.consolidateFiles 2 write in the last words Shuffle Summary of tuning Most of the performance of Spark operations is mainly consumed in the shuffle link, because the link contains a large number of disk IO, serialization, network data transmission and othe...
Lire la suite

Spark optimization

Image
One operation and maintenance 1. Master hang up, standby restart is also invalid Master defaults to 512M of memory, when the task in the cluster is particularly high, it will hang, because the master will read each task event log log to generate spark ui, the memory will naturally OOM, you can run the log See that the master of the start through the HA will naturally fail for this reason. solve Increase the Master's memory spark-env.sh , set in the master node spark-env.sh : export SPARK_DAEMON_MEMORY 10g # 根据你的实际情况 Reduce the job information stored in the Master memory spark.ui.retainedJobs 500 # 默认都是1000 spark.ui.retainedStages 500 Hang up or suspend Sometimes we will see the web node in the web ui disappear or in the dead state, the task of running the node will report a variety of lost worker errors, causing the same reasons and the above, worker memory to save a lot of ui The information leads to gc when the heartbeat is lost...
Lire la suite