http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b787e2f/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/apt/MapReduceTutorial.apt.vm ---------------------------------------------------------------------- diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/apt/MapReduceTutorial.apt.vm b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/apt/MapReduceTutorial.apt.vm deleted file mode 100644 index 9fb1056..0000000 --- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/apt/MapReduceTutorial.apt.vm +++ /dev/null @@ -1,1605 +0,0 @@ -~~ Licensed under the Apache License, Version 2.0 (the "License"); -~~ you may not use this file except in compliance with the License. -~~ You may obtain a copy of the License at -~~ -~~ http://www.apache.org/licenses/LICENSE-2.0 -~~ -~~ Unless required by applicable law or agreed to in writing, software -~~ distributed under the License is distributed on an "AS IS" BASIS, -~~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -~~ See the License for the specific language governing permissions and -~~ limitations under the License. See accompanying LICENSE file. - - --- - MapReduce Tutorial - --- - --- - ${maven.build.timestamp} - -MapReduce Tutorial - -%{toc|section=1|fromDepth=0|toDepth=4} - -* Purpose - - This document comprehensively describes all user-facing facets of - the Hadoop MapReduce framework and serves as a tutorial. - -* Prerequisites - - Ensure that Hadoop is installed, configured and is running. More details: - - * {{{../../hadoop-project-dist/hadoop-common/SingleCluster.html} - Single Node Setup}} for first-time users. - - * {{{../../hadoop-project-dist/hadoop-common/ClusterSetup.html} - Cluster Setup}} for large, distributed clusters. - -* Overview - - Hadoop MapReduce is a software framework for easily writing applications - which process vast amounts of data (multi-terabyte data-sets) in-parallel - on large clusters (thousands of nodes) of commodity hardware in a reliable, - fault-tolerant manner. - - A MapReduce <job> usually splits the input data-set into independent chunks - which are processed by the <map tasks> in a completely parallel manner. The - framework sorts the outputs of the maps, which are then input to the <reduce - tasks>. Typically both the input and the output of the job are stored in - a file-system. The framework takes care of scheduling tasks, monitoring them - and re-executes the failed tasks. - - Typically the compute nodes and the storage nodes are the same, that is, - the MapReduce framework and the Hadoop Distributed File System - (see {{{../../hadoop-project-dist/hadoop-hdfs/HdfsDesign.html} - HDFS Architecture Guide}}) are running on the same set of nodes. This - configuration allows the framework to effectively schedule tasks on the nodes - where data is already present, resulting in very high aggregate bandwidth - across the cluster. - - The MapReduce framework consists of a single master <<<ResourceManager>>>, - one slave <<<NodeManager>>> per cluster-node, and <<<MRAppMaster>>> per - application (see {{{../../hadoop-yarn/hadoop-yarn-site/YARN.html} - YARN Architecture Guide}}). - - Minimally, applications specify the input/output locations and supply <map> - and <reduce> functions via implementations of appropriate interfaces and/or - abstract-classes. These, and other job parameters, comprise the <job - configuration>. - - The Hadoop <job client> then submits the job (jar/executable etc.) and - configuration to the <<<ResourceManager>>> which then assumes the - responsibility of distributing the software/configuration to the slaves, - scheduling tasks and monitoring them, providing status and diagnostic - information to the job-client. - - Although the Hadoop framework is implemented in Java\u2122, MapReduce - applications need not be written in Java. - - * {{{../../api/org/apache/hadoop/streaming/package-summary.html} - Hadoop Streaming}} is a utility which allows users to create and run jobs - with any executables (e.g. shell utilities) as the mapper and/or the - reducer. - - * {{{../../api/org/apache/hadoop/mapred/pipes/package-summary.html} - Hadoop Pipes}} is a {{{http://www.swig.org/}SWIG}}-compatible C++ API to - implement MapReduce applications (non JNI\u2122 based). - -* Inputs and Outputs - - The MapReduce framework operates exclusively on <<<\<key, value\>>>> pairs, - that is, the framework views the input to the job as a set of <<<\<key, - value\>>>> pairs and produces a set of <<<\<key, value\>>>> pairs as the - output of the job, conceivably of different types. - - The <<<key>>> and <<<value>>> classes have to be serializable by the - framework and hence need to implement the - {{{../../api/org/apache/hadoop/io/Writable.html}Writable}} interface. - Additionally, the key classes have to implement the - {{{../../api/org/apache/hadoop/io/WritableComparable.html} - WritableComparable}} interface to facilitate sorting by the framework. - - Input and Output types of a MapReduce job: - - (input) <<<\<k1, v1\> -\>>>> <<map>> <<<-\> \<k2, v2\> -\>>>> <<combine>> - <<<-\> \<k2, v2\> -\>>>> <<reduce>> <<<-\> \<k3, v3\>>>> (output) - -* Example: WordCount v1.0 - - Before we jump into the details, lets walk through an example MapReduce - application to get a flavour for how they work. - - <<<WordCount>>> is a simple application that counts the number of - occurrences of each word in a given input set. - - This works with a local-standalone, pseudo-distributed or fully-distributed - Hadoop installation - ({{{../../hadoop-project-dist/hadoop-common/SingleCluster.html} - Single Node Setup}}). - -** Source Code - -+---+ -import java.io.IOException; -import java.util.StringTokenizer; - -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.io.IntWritable; -import org.apache.hadoop.io.Text; -import org.apache.hadoop.mapreduce.Job; -import org.apache.hadoop.mapreduce.Mapper; -import org.apache.hadoop.mapreduce.Reducer; -import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; -import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; - -public class WordCount { - - public static class TokenizerMapper - extends Mapper<Object, Text, Text, IntWritable>{ - - private final static IntWritable one = new IntWritable(1); - private Text word = new Text(); - - public void map(Object key, Text value, Context context - ) throws IOException, InterruptedException { - StringTokenizer itr = new StringTokenizer(value.toString()); - while (itr.hasMoreTokens()) { - word.set(itr.nextToken()); - context.write(word, one); - } - } - } - - public static class IntSumReducer - extends Reducer<Text,IntWritable,Text,IntWritable> { - private IntWritable result = new IntWritable(); - - public void reduce(Text key, Iterable<IntWritable> values, - Context context - ) throws IOException, InterruptedException { - int sum = 0; - for (IntWritable val : values) { - sum += val.get(); - } - result.set(sum); - context.write(key, result); - } - } - - public static void main(String[] args) throws Exception { - Configuration conf = new Configuration(); - Job job = Job.getInstance(conf, "word count"); - job.setJarByClass(WordCount.class); - job.setMapperClass(TokenizerMapper.class); - job.setCombinerClass(IntSumReducer.class); - job.setReducerClass(IntSumReducer.class); - job.setOutputKeyClass(Text.class); - job.setOutputValueClass(IntWritable.class); - FileInputFormat.addInputPath(job, new Path(args[0])); - FileOutputFormat.setOutputPath(job, new Path(args[1])); - System.exit(job.waitForCompletion(true) ? 0 : 1); - } -} -+---+ - -** Usage - - Assuming environment variables are set as follows: - -+---+ -export JAVA_HOME=/usr/java/default -export PATH=$JAVA_HOME/bin:$PATH -export HADOOP_CLASSPATH=$JAVA_HOME/lib/tools.jar -+---+ - - Compile <<<WordCount.java>>> and create a jar: - - <<<$ bin/hadoop com.sun.tools.javac.Main WordCount.java>>> \ - <<<$ jar cf wc.jar WordCount\*.class>>> - - Assuming that: - - * <<</user/joe/wordcount/input>>> - input directory in HDFS - - * <<</user/joe/wordcount/output>>> - output directory in HDFS - - Sample text-files as input: - - <<<$ bin/hdfs dfs -ls /user/joe/wordcount/input/>>> \ - <<</user/joe/wordcount/input/file01>>> \ - <<</user/joe/wordcount/input/file02>>> - - <<<$ bin/hdfs dfs -cat /user/joe/wordcount/input/file01>>> \ - <<<Hello World Bye World>>> - - <<<$ bin/hdfs dfs -cat /user/joe/wordcount/input/file02>>> \ - <<<Hello Hadoop Goodbye Hadoop>>> - - Run the application: - - <<<$ bin/hadoop jar wc.jar WordCount /user/joe/wordcount/input - /user/joe/wordcount/output>>> - - Output: - - <<<$ bin/hdfs dfs -cat /user/joe/wordcount/output/part-r-00000>>> - - <<<Bye 1>>> \ - <<<Goodbye 1>>> \ - <<<Hadoop 2>>> \ - <<<Hello 2>>> \ - <<<World 2>>> - - Applications can specify a comma separated list of paths which would be - present in the current working directory of the task using the option - <<<-files>>>. The <<<-libjars>>> option allows applications to add jars to - the classpaths of the maps and reduces. The option <<<-archives>>> allows - them to pass comma separated list of archives as arguments. These archives - are unarchived and a link with name of the archive is created in the current - working directory of tasks. More details about the command line options are - available at {{{../../hadoop-project-dist/hadoop-common/CommandsManual.html} - Commands Guide}}. - - Running <<<wordcount>>> example with <<<-libjars>>>, <<<-files>>> and - <<<-archives>>>: \ - <<<bin/hadoop jar hadoop-mapreduce-examples-<ver>.jar wordcount -files - cachefile.txt -libjars mylib.jar -archives myarchive.zip input output>>> - Here, myarchive.zip will be placed and unzipped into a directory by the name - "myarchive.zip". - - Users can specify a different symbolic name for files and archives passed - through <<<-files>>> and <<<-archives>>> option, using #. - - For example, <<<bin/hadoop jar hadoop-mapreduce-examples-<ver>.jar wordcount - -files dir1/dict.txt#dict1,dir2/dict.txt#dict2 -archives mytar.tgz#tgzdir - input output>>> Here, the files dir1/dict.txt and dir2/dict.txt can be - accessed by tasks using the symbolic names dict1 and dict2 respectively. - The archive mytar.tgz will be placed and unarchived into a directory by the - name "tgzdir". - -** Walk-through - - The <<<WordCount>>> application is quite straight-forward. - -+---+ - public void map(Object key, Text value, Context context - ) throws IOException, InterruptedException { - StringTokenizer itr = new StringTokenizer(value.toString()); - while (itr.hasMoreTokens()) { - word.set(itr.nextToken()); - context.write(word, one); - } - } -+---+ - - The <<<Mapper>>> implementation, via the <<<map>>> method, processes one - line at a time, as provided by the specified <<<TextInputFormat>>>. It then - splits the line into tokens separated by whitespaces, via the - <<<StringTokenizer>>>, and emits a key-value pair of <<<\< \<word\>, 1\>>>>. - - For the given sample input the first map emits: \ - <<<\< Hello, 1\>>>> \ - <<<\< World, 1\>>>> \ - <<<\< Bye, 1\>>>> \ - <<<\< World, 1\>>>> - - The second map emits: \ - <<<\< Hello, 1\>>>> \ - <<<\< Hadoop, 1\>>>> \ - <<<\< Goodbye, 1\>>>> \ - <<<\< Hadoop, 1\>>>> - - We'll learn more about the number of maps spawned for a given job, and how to - control them in a fine-grained manner, a bit later in the tutorial. - -+---+ - job.setCombinerClass(IntSumReducer.class); -+---+ - - <<<WordCount>>> also specifies a <<<combiner>>>. Hence, the output of each - map is passed through the local combiner (which is same as the <<<Reducer>>> - as per the job configuration) for local aggregation, after being sorted on - the <key>s. - - The output of the first map: \ - <<<\< Bye, 1\>>>> \ - <<<\< Hello, 1\>>>> \ - <<<\< World, 2\>>>> - - The output of the second map: \ - <<<\< Goodbye, 1\>>>> \ - <<<\< Hadoop, 2\>>>> \ - <<<\< Hello, 1\>>>> - -+---+ - public void reduce(Text key, Iterable<IntWritable> values, - Context context - ) throws IOException, InterruptedException { - int sum = 0; - for (IntWritable val : values) { - sum += val.get(); - } - result.set(sum); - context.write(key, result); - } -+---+ - - The <<<Reducer>>> implementation, via the <<<reduce>>> method just sums up - the values, which are the occurence counts for each key (i.e. words in this - example). - - Thus the output of the job is: \ - <<<\< Bye, 1\>>>> \ - <<<\< Goodbye, 1\>>>> \ - <<<\< Hadoop, 2\>>>> \ - <<<\< Hello, 2\>>>> \ - <<<\< World, 2\>>>> - - The <<<main>>> method specifies various facets of the job, such as the - input/output paths (passed via the command line), key/value types, - input/output formats etc., in the <<<Job>>>. It then calls the - <<<job.waitForCompletion>>> to submit the job and monitor its progress. - - We'll learn more about <<<Job>>>, <<<InputFormat>>>, <<<OutputFormat>>> and - other interfaces and classes a bit later in the tutorial. - -* MapReduce - User Interfaces - - This section provides a reasonable amount of detail on every user-facing - aspect of the MapReduce framework. This should help users implement, - configure and tune their jobs in a fine-grained manner. However, please note - that the javadoc for each class/interface remains the most comprehensive - documentation available; this is only meant to be a tutorial. - - Let us first take the <<<Mapper>>> and <<<Reducer>>> interfaces. Applications - typically implement them to provide the <<<map>>> and <<<reduce>>> methods. - - We will then discuss other core interfaces including <<<Job>>>, - <<<Partitioner>>>, <<<InputFormat>>>, <<<OutputFormat>>>, and others. - - Finally, we will wrap up by discussing some useful features of the framework - such as the <<<DistributedCache>>>, <<<IsolationRunner>>> etc. - -** Payload - - Applications typically implement the <<<Mapper>>> and <<<Reducer>>> - interfaces to provide the <<<map>>> and <<<reduce>>> methods. These form - the core of the job. - -*** Mapper - - {{{../../api/org/apache/hadoop/mapreduce/Mapper.html}Mapper}} maps input - key/value pairs to a set of intermediate key/value pairs. - - Maps are the individual tasks that transform input records into intermediate - records. The transformed intermediate records do not need to be of the same - type as the input records. A given input pair may map to zero or many output - pairs. - - The Hadoop MapReduce framework spawns one map task for each <<<InputSplit>>> - generated by the <<<InputFormat>>> for the job. - - Overall, <<<Mapper>>> implementations are passed the <<<Job>>> for the job - via the {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setMapperClass(Class)}} method. The framework then calls - {{{../../api/org/apache/hadoop/mapreduce/Mapper.html} - map(WritableComparable, Writable, Context)}} for each key/value pair in the - <<<InputSplit>>> for that task. Applications can then override the - <<<cleanup(Context)>>> method to perform any required cleanup. - - Output pairs do not need to be of the same types as input pairs. A given - input pair may map to zero or many output pairs. Output pairs are collected - with calls to context.write(WritableComparable, Writable). - - Applications can use the <<<Counter>>> to report its statistics. - - All intermediate values associated with a given output key are subsequently - grouped by the framework, and passed to the <<<Reducer>>>(s) to determine the - final output. Users can control the grouping by specifying a <<<Comparator>>> - via {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setGroupingComparatorClass(Class)}}. - - The <<<Mapper>>> outputs are sorted and then partitioned per <<<Reducer>>>. - The total number of partitions is the same as the number of reduce tasks for - the job. Users can control which keys (and hence records) go to which - <<<Reducer>>> by implementing a custom <<<Partitioner>>>. - - Users can optionally specify a <<<combiner>>>, via - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setCombinerClass(Class)}}, to perform local aggregation of the - intermediate outputs, which helps to cut down the amount of data transferred - from the <<<Mapper>>> to the <<<Reducer>>>. - - The intermediate, sorted outputs are always stored in a simple (key-len, key, - value-len, value) format. Applications can control if, and how, the - intermediate outputs are to be compressed and the - {{{../../api/org/apache/hadoop/io/compress/CompressionCodec.html} - CompressionCodec}} to be used via the <<<Configuration>>>. - -**** How Many Maps? - - The number of maps is usually driven by the total size of the inputs, that - is, the total number of blocks of the input files. - - The right level of parallelism for maps seems to be around 10-100 maps - per-node, although it has been set up to 300 maps for very cpu-light map - tasks. Task setup takes a while, so it is best if the maps take at least a - minute to execute. - - Thus, if you expect 10TB of input data and have a blocksize of <<<128MB>>>, - you'll end up with 82,000 maps, unless - Configuration.set(<<<MRJobConfig.NUM_MAPS>>>, int) (which only provides a - hint to the framework) is used to set it even higher. - -*** Reducer - - {{{../../api/org/apache/hadoop/mapreduce/Reducer.html}Reducer}} reduces a - set of intermediate values which share a key to a smaller set of values. - - The number of reduces for the job is set by the user via - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setNumReduceTasks(int)}}. - - Overall, <<<Reducer>>> implementations are passed the <<<Job>>> for the - job via the {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setReducerClass(Class)}} method and can override it to initialize - themselves. The framework then calls - {{{../../api/org/apache/hadoop/mapreduce/Reducer.html} - reduce(WritableComparable, Iterable\<Writable\>, Context)}} method for each - <<<\<key, (list of values)\>>>> pair in the grouped inputs. Applications can - then override the <<<cleanup(Context)>>> method to perform any required - cleanup. - - <<<Reducer>>> has 3 primary phases: shuffle, sort and reduce. - -**** Shuffle - - Input to the <<<Reducer>>> is the sorted output of the mappers. In this phase - the framework fetches the relevant partition of the output of all the - mappers, via HTTP. - -**** Sort - - The framework groups <<<Reducer>>> inputs by keys (since different mappers - may have output the same key) in this stage. - - The shuffle and sort phases occur simultaneously; while map-outputs are being - fetched they are merged. - -**** Secondary Sort - - If equivalence rules for grouping the intermediate keys are required to be - different from those for grouping keys before reduction, then one may specify - a <<<Comparator>>> via - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setSortComparatorClass(Class)}}. Since - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setGroupingComparatorClass(Class)}} can be used to control how - intermediate keys are grouped, these can be used in conjunction to simulate - <secondary sort on values>. - -**** Reduce - - In this phase the reduce(WritableComparable, Iterable\<Writable\>, Context) - method is called for each <<<\<key, (list of values)\>>>> pair in the grouped - inputs. - - The output of the reduce task is typically written to the - {{{../../api/org/apache/hadoop/fs/FileSystem.html}FileSystem}} via - Context.write(WritableComparable, Writable). - - Applications can use the <<<Counter>>> to report its statistics. - - The output of the <<<Reducer>>> is <not sorted>. - -**** How Many Reduces? - - The right number of reduces seems to be <<<0.95>>> or <<<1.75>>> multiplied - by (\<<no. of nodes>\> * \<<no. of maximum containers per node>\>). - - With <<<0.95>>> all of the reduces can launch immediately and start - transferring map outputs as the maps finish. With <<<1.75>>> the faster nodes - will finish their first round of reduces and launch a second wave of reduces - doing a much better job of load balancing. - - Increasing the number of reduces increases the framework overhead, but - increases load balancing and lowers the cost of failures. - - The scaling factors above are slightly less than whole numbers to reserve a - few reduce slots in the framework for speculative-tasks and failed tasks. - -**** Reducer NONE - - It is legal to set the number of reduce-tasks to <zero> if no reduction is - desired. - - In this case the outputs of the map-tasks go directly to the - <<<FileSystem>>>, into the output path set by - {{{../../api/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.html} - FileOutputFormat.setOutputPath(Job, Path)}}. The framework does not sort the - map-outputs before writing them out to the <<<FileSystem>>>. - -*** Partitioner - - {{{../../api/org/apache/hadoop/mapreduce/Partitioner.html}Partitioner}} - partitions the key space. - - Partitioner controls the partitioning of the keys of the intermediate - map-outputs. The key (or a subset of the key) is used to derive the - partition, typically by a <hash function>. The total number of partitions is - the same as the number of reduce tasks for the job. Hence this controls which - of the <<<m>>> reduce tasks the intermediate key (and hence the record) is - sent to for reduction. - - {{{../../api/org/apache/hadoop/mapreduce/lib/partition/HashPartitioner.html} - HashPartitioner}} is the default <<<Partitioner>>>. - -*** Counter - - {{{../../api/org/apache/hadoop/mapreduce/Counter.html}Counter}} is a facility - for MapReduce applications to report its statistics. - - <<<Mapper>>> and <<<Reducer>>> implementations can use the <<<Counter>>> to - report statistics. - - Hadoop MapReduce comes bundled with a - {{{../../api/org/apache/hadoop/mapreduce/package-summary.html}library}} - of generally useful mappers, reducers, and partitioners. - -** Job Configuration - - {{{../../api/org/apache/hadoop/mapreduce/Job.html}Job}} represents a - MapReduce job configuration. - - <<<Job>>> is the primary interface for a user to describe a MapReduce job to - the Hadoop framework for execution. The framework tries to faithfully execute - the job as described by <<<Job>>>, however: - - * Some configuration parameters may have been marked as final by - administrators - (see {{{../../api/org/apache/hadoop/conf/Configuration.html#FinalParams} - Final Parameters}}) and hence cannot be altered. - - * While some job parameters are straight-forward to set (e.g. - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setNumReduceTasks(int)}}), other parameters interact subtly with the - rest of the framework and/or job configuration and are more complex to set - (e.g. {{{../../api/org/apache/hadoop/conf/Configuration.html} - Configuration.set(<<<JobContext.NUM_MAPS>>>, int)}}). - - <<<Job>>> is typically used to specify the <<<Mapper>>>, combiner (if any), - <<<Partitioner>>>, <<<Reducer>>>, <<<InputFormat>>>, <<<OutputFormat>>> - implementations. - {{{../../api/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.html} - FileInputFormat}} indicates the set of input files - ({{{../../api/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.html} - FileInputFormat.setInputPaths(Job, Path...)}}/ - {{{../../api/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.html} - FileInputFormat.addInputPath(Job, Path)}}) and - ({{{../../api/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.html} - FileInputFormat.setInputPaths(Job, String...)}}/ - {{{../../api/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.html} - FileInputFormat.addInputPaths(Job, String))}} and where the output files - should be written - ({{{../../api/org/apache/hadoop/mapreduce/lib/input/FileOutputFormat.html} - FileOutputFormat.setOutputPath(Path)}}). - - Optionally, <<<Job>>> is used to specify other advanced facets of the job - such as the <<<Comparator>>> to be used, files to be put in the - <<<DistributedCache>>>, whether intermediate and/or job outputs are to be - compressed (and how), whether job tasks can be executed in a <speculative> - manner ({{{../../api/org/apache/hadoop/mapreduce/Job.html} - setMapSpeculativeExecution(boolean)}})/ - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - setReduceSpeculativeExecution(boolean)}}), - maximum number of attempts per task - ({{{../../api/org/apache/hadoop/mapreduce/Job.html}setMaxMapAttempts(int)}}/ - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - setMaxReduceAttempts(int)}}) etc. - - Of course, users can use - {{{../../api/org/apache/hadoop/conf/Configuration.html} - Configuration.set(String, String)}}/ - {{{../../api/org/apache/hadoop/conf/Configuration.html} - Configuration.get(String)}} to set/get arbitrary parameters needed by - applications. However, use the <<<DistributedCache>>> for large amounts of - (read-only) data. - -** Task Execution & Environment - - The <<<MRAppMaster>>> executes the <<<Mapper>>>/<<<Reducer>>> <task> as a - child process in a separate jvm. - - The child-task inherits the environment of the parent <<<MRAppMaster>>>. The - user can specify additional options to the child-jvm via the - <<<mapreduce.\{map|reduce\}.java.opts>>> and configuration parameter in the - <<<Job>>> such as non-standard paths for the run-time linker to search - shared libraries via <<<-Djava.library.path=\<\>>>> etc. If the - <<<mapreduce.\{map|reduce\}.java.opts>>> parameters contains the symbol - <@taskid@> it is interpolated with value of <<<taskid>>> of the MapReduce - task. - - Here is an example with multiple arguments and substitutions, showing jvm GC - logging, and start of a passwordless JVM JMX agent so that it can connect - with jconsole and the likes to watch child memory, threads and get thread - dumps. It also sets the maximum heap-size of the map and reduce child jvm to - 512MB & 1024MB respectively. It also adds an additional path to the - <<<java.library.path>>> of the child-jvm. - -+---+ -<property> - <name>mapreduce.map.java.opts</name> - <value> - -Xmx512M -Djava.library.path=/home/mycompany/lib -verbose:gc -Xloggc:/tmp/@taskid@.gc - -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false - </value> -</property> - -<property> - <name>mapreduce.reduce.java.opts</name> - <value> - -Xmx1024M -Djava.library.path=/home/mycompany/lib -verbose:gc -Xloggc:/tmp/@taskid@.gc - -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false - </value> -</property> -+---+ - -*** Memory Management - - Users/admins can also specify the maximum virtual memory of the launched - child-task, and any sub-process it launches recursively, using - <<<mapreduce.\{map|reduce\}.memory.mb>>>. Note that the value set here is a - per process limit. The value for <<<mapreduce.\{map|reduce\}.memory.mb>>> - should be specified in mega bytes (MB). And also the value must be greater - than or equal to the -Xmx passed to JavaVM, else the VM might not start. - - Note: <<<mapreduce.\{map|reduce\}.java.opts>>> are used only for configuring - the launched child tasks from MRAppMaster. Configuring the memory options for - daemons is documented in - {{{../../hadoop-project-dist/hadoop-common/ClusterSetup.html#Configuring_Environment_of_Hadoop_Daemons} - Configuring the Environment of the Hadoop Daemons}}. - - The memory available to some parts of the framework is also configurable. - In map and reduce tasks, performance may be influenced by adjusting - parameters influencing the concurrency of operations and the frequency with - which data will hit disk. Monitoring the filesystem counters for a job- - particularly relative to byte counts from the map and into the reduce- is - invaluable to the tuning of these parameters. - -*** Map Parameters - - A record emitted from a map will be serialized into a buffer and metadata - will be stored into accounting buffers. As described in the following - options, when either the serialization buffer or the metadata exceed a - threshold, the contents of the buffers will be sorted and written to disk in - the background while the map continues to output records. If either buffer - fills completely while the spill is in progress, the map thread will block. - When the map is finished, any remaining records are written to disk and all - on-disk segments are merged into a single file. Minimizing the number of - spills to disk can decrease map time, but a larger buffer also decreases the - memory available to the mapper. - -*-------------*-------*-------------------------------------------------------* -|| Name || Type || Description | -*-------------+-------+-------------------------------------------------------+ -| mapreduce.task.io.sort.mb | int | The cumulative size of the serialization -| | | and accounting buffers storing records emitted from the -| | | map, in megabytes. -*-------------+-------+-------------------------------------------------------+ -| mapreduce.map.sort.spill.percent | float | The soft limit in the -| | | serialization buffer. Once reached, a thread will begin -| | | to spill the contents to disk in the background. -*-------------+-------+-------------------------------------------------------+ - - Other notes - - * If either spill threshold is exceeded while a spill is in progress, - collection will continue until the spill is finished. For example, if - <<<mapreduce.map.sort.spill.percent>>> is set to 0.33, and the remainder - of the buffer is filled while the spill runs, the next spill will include - all the collected records, or 0.66 of the buffer, and will not generate - additional spills. In other words, the thresholds are defining triggers, - not blocking. - - * A record larger than the serialization buffer will first trigger a spill, - then be spilled to a separate file. It is undefined whether or not this - record will first pass through the combiner. - -*** Shuffle/Reduce Parameters - - As described previously, each reduce fetches the output assigned to it by the - Partitioner via HTTP into memory and periodically merges these outputs to - disk. If intermediate compression of map outputs is turned on, each output is - decompressed into memory. The following options affect the frequency of these - merges to disk prior to the reduce and the memory allocated to map output - during the reduce. - -*-------------*-------*-------------------------------------------------------* -|| Name || Type || Description | -*-------------+-------+-------------------------------------------------------+ -| mapreduce.task.io.soft.factor | int | Specifies the number of segments on -| | | disk to be merged at the same time. It limits the -| | | number of open files and compression codecs during -| | | merge. If the number of files exceeds this limit, the -| | | merge will proceed in several passes. Though this limit -| | | also applies to the map, most jobs should be configured -| | | so that hitting this limit is unlikely there. -*-------------+-------+-------------------------------------------------------+ -| mapreduce.reduce.merge.inmem.thresholds | int | The number of sorted map -| | | outputs fetched into memory before being merged to -| | | disk. Like the spill thresholds in the preceding note, -| | | this is not defining a unit of partition, but a -| | | trigger. In practice, this is usually set very high -| | | (1000) or disabled (0), since merging in-memory -| | | segments is often less expensive than merging from disk -| | | (see notes following this table). This threshold -| | | influences only the frequency of in-memory merges -| | | during the shuffle. -*-------------+-------+-------------------------------------------------------+ -| mapreduce.reduce.shuffle.merge.percent | float | The memory threshold for -| | | fetched map outputs before an in-memory merge is started, -| | | expressed as a percentage of memory allocated to -| | | storing map outputs in memory. Since map outputs that -| | | can't fit in memory can be stalled, setting this high -| | | may decrease parallelism between the fetch and merge. -| | | Conversely, values as high as 1.0 have been effective -| | | for reduces whose input can fit entirely in memory. -| | | This parameter influences only the frequency of -| | | in-memory merges during the shuffle. -*-------------+-------+-------------------------------------------------------+ -| mapreduce.reduce.shuffle.input.buffer.percent | float | The percentage of -| | | memory- relative to the maximum heapsize as typically -| | | specified in <<<mapreduce.reduce.java.opts>>>- that can -| | | be allocated to storing map outputs during the shuffle. -| | | Though some memory should be set aside for the -| | | framework, in general it is advantageous to set this -| | | high enough to store large and numerous map outputs. -*-------------+-------+-------------------------------------------------------+ -| mapreduce.reduce.input.buffer.percent | float | The percentage of memory -| | | relative to the maximum heapsize in which map outputs -| | | may be retained during the reduce. When the reduce -| | | begins, map outputs will be merged to disk until those -| | | that remain are under the resource limit this defines. -| | | By default, all map outputs are merged to disk before -| | | the reduce begins to maximize the memory available to -| | | the reduce. For less memory-intensive reduces, this -| | | should be increased to avoid trips to disk. -*-------------+-------+-------------------------------------------------------+ - - Other notes - - * If a map output is larger than 25 percent of the memory allocated to - copying map outputs, it will be written directly to disk without first - staging through memory. - - * When running with a combiner, the reasoning about high merge thresholds - and large buffers may not hold. For merges started before all map outputs - have been fetched, the combiner is run while spilling to disk. In some - cases, one can obtain better reduce times by spending resources combining - map outputs- making disk spills small and parallelizing spilling and - fetching- rather than aggressively increasing buffer sizes. - - * When merging in-memory map outputs to disk to begin the reduce, if an - intermediate merge is necessary because there are segments to spill and at - least <<<mapreduce.task.io.sort.factor>>> segments already on disk, the - in-memory map outputs will be part of the intermediate merge. - -*** Configured Parameters - - The following properties are localized in the job configuration for each - task's execution: - -*-------------*-------*-------------------------------------------------------* -|| Name || Type || Description | -*-------------+-------+-------------------------------------------------------+ -| mapreduce.job.id | String | The job id -*-------------+-------+-------------------------------------------------------+ -| mapreduce.job.jar | String | job.jar location in job directory -*-------------+-------+-------------------------------------------------------+ -| mapreduce.job.local.dir | String | The job specific shared scratch space -*-------------+-------+-------------------------------------------------------+ -| mapreduce.task.id | String | The task id -*-------------+-------+-------------------------------------------------------+ -| mapreduce.task.attempt.id | String | The task attempt id -*-------------+-------+-------------------------------------------------------+ -| mapreduce.task.is.map | boolean | Is this a map task -*-------------+-------+-------------------------------------------------------+ -| mapreduce.task.partition | int | The id of the task within the job -*-------------+-------+-------------------------------------------------------+ -| mapreduce.map.input.file | String | The filename that the map is reading from -*-------------+-------+-------------------------------------------------------+ -| mapreduce.map.input.start | long | The offset of the start of the map input -| | | split -*-------------+-------+-------------------------------------------------------+ -| mapreduce.map.input.length | long | The number of bytes in the map input -| | | split -*-------------+-------+-------------------------------------------------------+ -| mapreduce.task.output.dir | String | The task's temporary output directory -*-------------+-------+-------------------------------------------------------+ - - <<Note:>> During the execution of a streaming job, the names of the - "mapreduce" parameters are transformed. The dots ( . ) become underscores - ( _ ). For example, mapreduce.job.id becomes mapreduce_job_id and - mapreduce.job.jar becomes mapreduce_job_jar. To get the values in a streaming - job's mapper/reducer use the parameter names with the underscores. - -*** Task Logs - - The standard output (stdout) and error (stderr) streams and the syslog of the - task are read by the NodeManager and logged to - <<<$\{HADOOP_LOG_DIR\}/userlogs>>>. - -*** Distributing Libraries - - The {{DistributedCache}} can also be used to distribute both jars and native - libraries for use in the map and/or reduce tasks. The child-jvm always has - its <current working directory> added to the <<<java.library.path>>> and - <<<LD_LIBRARY_PATH>>>. And hence the cached libraries can be loaded via - {{{http://docs.oracle.com/javase/7/docs/api/java/lang/System.html} - System.loadLibrary}} or - {{{http://docs.oracle.com/javase/7/docs/api/java/lang/System.html} - System.load}}. More details on how to load shared libraries through - distributed cache are documented at - {{{../../hadoop-project-dist/hadoop-common/NativeLibraries.html#Native_Shared_Libraries} - Native Libraries}}. - -** Job Submission and Monitoring - - {{{../../api/org/apache/hadoop/mapreduce/Job.html}Job}} is the primary - interface by which user-job interacts with the <<<ResourceManager>>>. - - <<<Job>>> provides facilities to submit jobs, track their progress, access - component-tasks' reports and logs, get the MapReduce cluster's status - information and so on. - - The job submission process involves: - - [[1]] Checking the input and output specifications of the job. - - [[2]] Computing the <<<InputSplit>>> values for the job. - - [[3]] Setting up the requisite accounting information for the - <<<DistributedCache>>> of the job, if necessary. - - [[4]] Copying the job's jar and configuration to the MapReduce system - directory on the <<<FileSystem>>>. - - [[5]] Submitting the job to the <<<ResourceManager>>> and optionally - monitoring it's status. - - Job history files are also logged to user specified directory - <<<mapreduce.jobhistory.intermediate-done-dir>>> and - <<<mapreduce.jobhistory.done-dir>>>, which defaults to job output directory. - - User can view the history logs summary in specified directory using the - following command \ - <<<$ mapred job -history output.jhist>>> \ - This command will print job details, failed and killed tip details. \ - More details about the job such as successful tasks and task attempts made - for each task can be viewed using the following command \ - <<<$ mapred job -history all output.jhist>>> - - Normally the user uses <<<Job>>> to create the application, describe various - facets of the job, submit the job, and monitor its progress. - -*** Job Control - - Users may need to chain MapReduce jobs to accomplish complex tasks which - cannot be done via a single MapReduce job. This is fairly easy since the - output of the job typically goes to distributed file-system, and the output, - in turn, can be used as the input for the next job. - - However, this also means that the onus on ensuring jobs are complete - (success/failure) lies squarely on the clients. In such cases, the various - job-control options are: - - * {{{../../api/org/apache/hadoop/mapreduce/Job.html}Job.submit()}} : - Submit the job to the cluster and return immediately. - - * {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.waitForCompletion(boolean)}} : - Submit the job to the cluster and wait for it to finish. - -** Job Input - - {{{../../api/org/apache/hadoop/mapreduce/InputFormat.html}InputFormat}} - describes the input-specification for a MapReduce job. - - The MapReduce framework relies on the <<<InputFormat>>> of the job to: - - [[1]] Validate the input-specification of the job. - - [[2]] Split-up the input file(s) into logical <<<InputSplit>>> instances, - each of which is then assigned to an individual <<<Mapper>>>. - - [[3]] Provide the <<<RecordReader>>> implementation used to glean input - records from the logical <<<InputSplit>>> for processing by the - <<<Mapper>>>. - - The default behavior of file-based <<<InputFormat>>> implementations, - typically sub-classes of - {{{../../api/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.html} - FileInputFormat}}, is to split the input into <logical> <<<InputSplit>>> - instances based on the total size, in bytes, of the input files. However, the - <<<FileSystem>>> blocksize of the input files is treated as an upper bound - for input splits. A lower bound on the split size can be set via - <<<mapreduce.input.fileinputformat.split.minsize>>>. - - Clearly, logical splits based on input-size is insufficient for many - applications since record boundaries must be respected. In such cases, the - application should implement a <<<RecordReader>>>, who is responsible for - respecting record-boundaries and presents a record-oriented view of the - logical <<<InputSplit>>> to the individual task. - - {{{../../api/org/apache/hadoop/mapreduce/lib/input/TextInputFormat.html} - TextInputFormat}} is the default <<<InputFormat>>>. - - If <<<TextInputFormat>>> is the <<<InputFormat>>> for a given job, the - framework detects input-files with the <.gz> extensions and automatically - decompresses them using the appropriate <<<CompressionCodec>>>. However, it - must be noted that compressed files with the above extensions cannot be - <split> and each compressed file is processed in its entirety by a single - mapper. - -*** InputSplit - - {{{../../api/org/apache/hadoop/mapreduce/InputSplit.html}InputSplit}} - represents the data to be processed by an individual <<<Mapper>>>. - - Typically <<<InputSplit>>> presents a byte-oriented view of the input, and it - is the responsibility of <<<RecordReader>>> to process and present a - record-oriented view. - - {{{../../api/org/apache/hadoop/mapreduce/lib/input/FileSplit.html}FileSplit}} - is the default <<<InputSplit>>>. It sets <<<mapreduce.map.input.file>>> to - the path of the input file for the logical split. - -*** RecordReader - - {{{../../api/org/apache/hadoop/mapreduce/RecordReader.html}RecordReader}} - reads <<<\<key, value\>>>> pairs from an <<<InputSplit>>>. - - Typically the <<<RecordReader>>> converts the byte-oriented view of the - input, provided by the <<<InputSplit>>>, and presents a record-oriented to - the <<<Mapper>>> implementations for processing. <<<RecordReader>>> thus - assumes the responsibility of processing record boundaries and presents the - tasks with keys and values. - -** Job Output - - {{{../../api/org/apache/hadoop/mapreduce/OutputFormat.html}OutputFormat}} - describes the output-specification for a MapReduce job. - - The MapReduce framework relies on the <<<OutputFormat>>> of the job to: - - [[1]] Validate the output-specification of the job; for example, check that - the output directory doesn't already exist. - - [[2]] Provide the <<<RecordWriter>>> implementation used to write the output - files of the job. Output files are stored in a <<<FileSystem>>>. - - <<<TextOutputFormat>>> is the default <<<OutputFormat>>>. - -*** OutputCommitter - - {{{../../api/org/apache/hadoop/mapreduce/OutputCommitter.html} - OutputCommitter}} describes the commit of task output for a MapReduce job. - - The MapReduce framework relies on the <<<OutputCommitter>>> of the job to: - - [[1]] Setup the job during initialization. For example, create the temporary - output directory for the job during the initialization of the job. Job - setup is done by a separate task when the job is in PREP state and - after initializing tasks. Once the setup task completes, the job will - be moved to RUNNING state. - - [[2]] Cleanup the job after the job completion. For example, remove the - temporary output directory after the job completion. Job cleanup is - done by a separate task at the end of the job. Job is declared - SUCCEDED/FAILED/KILLED after the cleanup task completes. - - [[3]] Setup the task temporary output. Task setup is done as part of the - same task, during task initialization. - - [[4]] Check whether a task needs a commit. This is to avoid the commit - procedure if a task does not need commit. - - [[5]] Commit of the task output. Once task is done, the task will commit - it's output if required. - - [[6]] Discard the task commit. If the task has been failed/killed, the - output will be cleaned-up. If task could not cleanup (in exception - block), a separate task will be launched with same attempt-id to do - the cleanup. - - <<<FileOutputCommitter>>> is the default <<<OutputCommitter>>>. Job - setup/cleanup tasks occupy map or reduce containers, whichever is available - on the NodeManager. And JobCleanup task, TaskCleanup tasks and JobSetup task - have the highest priority, and in that order. - -*** Task Side-Effect Files - - In some applications, component tasks need to create and/or write to - side-files, which differ from the actual job-output files. - - In such cases there could be issues with two instances of the same - <<<Mapper>>> or <<<Reducer>>> running simultaneously (for example, - speculative tasks) trying to open and/or write to the same file (path) on the - <<<FileSystem>>>. Hence the application-writer will have to pick unique names - per task-attempt (using the attemptid, say - <<<attempt_200709221812_0001_m_000000_0>>>), not just per task. - - To avoid these issues the MapReduce framework, when the <<<OutputCommitter>>> - is <<<FileOutputCommitter>>>, maintains a special - <<<$\{mapreduce.output.fileoutputformat.outputdir\}/_temporary/_$\{taskid\}>>> - sub-directory accessible via <<<$\{mapreduce.task.output.dir\}>>> for each - task-attempt on the <<<FileSystem>>> where the output of the task-attempt is - stored. On successful completion of the task-attempt, the files in the - <<<$\{mapreduce.output.fileoutputformat.outputdir\}/_temporary/_$\{taskid\}>>> - (only) are <promoted> to - <<<$\{mapreduce.output.fileoutputformat.outputdir\}>>>. Of course, the - framework discards the sub-directory of unsuccessful task-attempts. This - process is completely transparent to the application. - - The application-writer can take advantage of this feature by creating any - side-files required in <<<$\{mapreduce.task.output.dir\}>>> during execution - of a task via - {{{../../api/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.html} - FileOutputFormat.getWorkOutputPath(Conext)}}, and the framework will promote - them similarly for succesful task-attempts, thus eliminating the need to pick - unique paths per task-attempt. - - Note: The value of <<<$\{mapreduce.task.output.dir\}>>> during execution of a - particular task-attempt is actually - <<<$\{mapreduce.output.fileoutputformat.outputdir\}/_temporary/_\{$taskid\}>>>, - and this value is set by the MapReduce framework. So, just create any - side-files in the path returned by - {{{../../api/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.html} - FileOutputFormat.getWorkOutputPath(Conext)}} from MapReduce task to take - advantage of this feature. - - The entire discussion holds true for maps of jobs with reducer=NONE - (i.e. 0 reduces) since output of the map, in that case, goes directly to - HDFS. - -*** RecordWriter - - {{{../../api/org/apache/hadoop/mapreduce/RecordWriter.html}RecordWriter}} - writes the output <<<\<key, value\>>>> pairs to an output file. - - RecordWriter implementations write the job outputs to the <<<FileSystem>>>. - -** Other Useful Features - -*** Submitting Jobs to Queues - - Users submit jobs to Queues. Queues, as collection of jobs, allow the system - to provide specific functionality. For example, queues use ACLs to control - which users who can submit jobs to them. Queues are expected to be primarily - used by Hadoop Schedulers. - - Hadoop comes configured with a single mandatory queue, called 'default'. - Queue names are defined in the <<<mapreduce.job.queuename>>>> property of the - Hadoop site configuration. Some job schedulers, such as the - {{{../../hadoop-yarn/hadoop-yarn-site/CapacityScheduler.html} - Capacity Scheduler}}, support multiple queues. - - A job defines the queue it needs to be submitted to through the - <<<mapreduce.job.queuename>>> property, or through the - Configuration.set(<<<MRJobConfig.QUEUE_NAME>>>, String) API. Setting the - queue name is optional. If a job is submitted without an associated queue - name, it is submitted to the 'default' queue. - -*** Counters - - <<<Counters>>> represent global counters, defined either by the MapReduce - framework or applications. Each <<<Counter>>> can be of any <<<Enum>>> type. - Counters of a particular <<<Enum>>> are bunched into groups of type - <<<Counters.Group>>>. - - Applications can define arbitrary <<<Counters>>> (of type <<<Enum>>>) and - update them via - {{{../../api/org/apache/hadoop/mapred/Counters.html} - Counters.incrCounter(Enum, long)}} or Counters.incrCounter(String, String, - long) in the <<<map>>> and/or <<<reduce>>> methods. These counters are then - globally aggregated by the framework. - -*** DistributedCache - - <<<DistributedCache>>> distributes application-specific, large, read-only - files efficiently. - - <<<DistributedCache>>> is a facility provided by the MapReduce framework to - cache files (text, archives, jars and so on) needed by applications. - - Applications specify the files to be cached via urls (hdfs://) in the - <<<Job>>>. The <<<DistributedCache>>> assumes that the files specified via - hdfs:// urls are already present on the <<<FileSystem>>>. - - The framework will copy the necessary files to the slave node before any - tasks for the job are executed on that node. Its efficiency stems from the - fact that the files are only copied once per job and the ability to cache - archives which are un-archived on the slaves. - - <<<DistributedCache>>> tracks the modification timestamps of the cached - files. Clearly the cache files should not be modified by the application or - externally while the job is executing. - - <<<DistributedCache>>> can be used to distribute simple, read-only data/text - files and more complex types such as archives and jars. Archives (zip, tar, - tgz and tar.gz files) are <un-archived> at the slave nodes. Files have - <execution permissions> set. - - The files/archives can be distributed by setting the property - <<<mapreduce.job.cache.\{files|archives\}>>>. If more than one file/archive - has to be distributed, they can be added as comma separated paths. The - properties can also be set by APIs - {{{../../api/org/apache/hadoop/mapreduce/Job.html}Job.addCacheFile(URI)}}/ - {{{../../api/org/apache/hadoop/mapreduce/Job.html}Job.addCacheArchive(URI)}} - and - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setCacheFiles(URI\[\])}}/ - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setCacheArchives(URI\[\])}} where URI is of the form - <<<hdfs://host:port/absolute-path\#link-name>>>. In Streaming, the files can - be distributed through command line option <<<-cacheFile/-cacheArchive>>>. - - The <<<DistributedCache>>> can also be used as a rudimentary software - distribution mechanism for use in the map and/or reduce tasks. It can be used - to distribute both jars and native libraries. The - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.addArchiveToClassPath(Path)}} or - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.addFileToClassPath(Path)}} api can be used to cache files/jars and also - add them to the <classpath> of child-jvm. The same can be done by setting the - configuration properties <<<mapreduce.job.classpath.\{files|archives\}>>>. - Similarly the cached files that are symlinked into the working directory of - the task can be used to distribute native libraries and load them. - -**** Private and Public DistributedCache Files - - DistributedCache files can be private or public, that determines how they can - be shared on the slave nodes. - - * "Private" DistributedCache files are cached in a localdirectory private to - the user whose jobs need these files. These files are shared by all tasks - and jobs of the specific user only and cannot be accessed by jobs of - other users on the slaves. A DistributedCache file becomes private by - virtue of its permissions on the file system where the files are - uploaded, typically HDFS. If the file has no world readable access, or if - the directory path leading to the file has no world executable access for - lookup, then the file becomes private. - - * "Public" DistributedCache files are cached in a global directory and the - file access is setup such that they are publicly visible to all users. - These files can be shared by tasks and jobs of all users on the slaves. A - DistributedCache file becomes public by virtue of its permissions on the - file system where the files are uploaded, typically HDFS. If the file has - world readable access, AND if the directory path leading to the file has - world executable access for lookup, then the file becomes public. In other - words, if the user intends to make a file publicly available to all users, - the file permissions must be set to be world readable, and the directory - permissions on the path leading to the file must be world executable. - -*** Profiling - - Profiling is a utility to get a representative (2 or 3) sample of built-in - java profiler for a sample of maps and reduces. - - User can specify whether the system should collect profiler information for - some of the tasks in the job by setting the configuration property - <<<mapreduce.task.profile>>>. The value can be set using the api - Configuration.set(<<<MRJobConfig.TASK_PROFILE>>>, boolean). If the value is - set <<<true>>>, the task profiling is enabled. The profiler information is - stored in the user log directory. By default, profiling is not enabled for - the job. - - Once user configures that profiling is needed, she/he can use the - configuration property <<<mapreduce.task.profile.\{maps|reduces\}>>> - to set the ranges of MapReduce tasks to profile. The value can be set using - the api Configuration.set(<<<MRJobConfig.NUM_\{MAP|REDUCE\}_PROFILES>>>, - String). By default, the specified range is <<<0-2>>>. - - User can also specify the profiler configuration arguments by setting the - configuration property <<<mapreduce.task.profile.params>>>. The value can be - specified using the api - Configuration.set(<<<MRJobConfig.TASK_PROFILE_PARAMS>>>, String). If the - string contains a <<<%s>>>, it will be replaced with the name of the - profiling output file when the task runs. These parameters are passed to the - task child JVM on the command line. The default value for the profiling - parameters is - <<<-agentlib:hprof=cpu=samples,heap=sites,force=n,thread=y,verbose=n,file=%s>>>. - -*** Debugging - - The MapReduce framework provides a facility to run user-provided scripts for - debugging. When a MapReduce task fails, a user can run a debug script, to - process task logs for example. The script is given access to the task's - stdout and stderr outputs, syslog and jobconf. The output from the debug - script's stdout and stderr is displayed on the console diagnostics and also - as part of the job UI. - - In the following sections we discuss how to submit a debug script with a job. - The script file needs to be distributed and submitted to the framework. - -**** How to distribute the script file: - - The user needs to use {{DistributedCache}} to <distribute> and <symlink> the - script file. - -**** How to submit the script: - - A quick way to submit the debug script is to set values for the properties - <<<mapreduce.map.debug.script>>> and <<<mapreduce.reduce.debug.script>>>, for - debugging map and reduce tasks respectively. These properties can also be set - by using APIs - {{{../../api/org/apache/hadoop/conf/Configuration.html} - Configuration.set(<<<MRJobConfig.MAP_DEBUG_SCRIPT>>>, String)}} and - {{{../../api/org/apache/hadoop/conf/Configuration.html} - Configuration.set(<<<MRJobConfig.REDUCE_DEBUG_SCRIPT>>>, String)}}. In - streaming mode, a debug script can be submitted with the command-line options - <<<-mapdebug>>> and <<<-reducedebug>>>, for debugging map and reduce tasks - respectively. - - The arguments to the script are the task's stdout, stderr, syslog and jobconf - files. The debug command, run on the node where the MapReduce task failed, - is: \ - <<<$script $stdout $stderr $syslog $jobconf>>> - - Pipes programs have the c++ program name as a fifth argument for the command. - Thus for the pipes programs the command is \ - <<<$script $stdout $stderr $syslog $jobconf $program>>> - -**** Default Behavior: - - For pipes, a default script is run to process core dumps under gdb, prints - stack trace and gives info about running threads. - -*** Data Compression - - Hadoop MapReduce provides facilities for the application-writer to specify - compression for both intermediate map-outputs and the job-outputs i.e. output - of the reduces. It also comes bundled with - {{{../../api/org/apache/hadoop/io/compress/CompressionCodec.html} - CompressionCodec}} implementation for the {{{http://www.zlib.net}zlib}} - compression algorithm. The {{{http://www.gzip.org}gzip}}, - {{{http://www.bzip.org}bzip2}}, {{{http://code.google.com/p/snappy/}snappy}}, - and {{{http://code.google.com/p/lz4/}lz4}} file format are also supported. - - Hadoop also provides native implementations of the above compression codecs - for reasons of both performance (zlib) and non-availability of Java - libraries. More details on their usage and availability are available - {{{../../hadoop-project-dist/hadoop-common/NativeLibraries.html}here}}. - -**** Intermediate Outputs - - Applications can control compression of intermediate map-outputs via the - Configuration.set(<<<MRJobConfig.MAP_OUTPUT_COMPRESS>>>, boolean) api and the - <<<CompressionCodec>>> to be used via the - Configuration.set(<<<MRJobConfig.MAP_OUTPUT_COMPRESS_CODEC>>>, Class) api. - -**** Job Outputs - - Applications can control compression of job-outputs via the - {{{../../api/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.html} - FileOutputFormat.setCompressOutput(Job, boolean)}} api and the - <<<CompressionCodec>>> to be used can be specified via the - FileOutputFormat.setOutputCompressorClass(Job, Class) api. - - If the job outputs are to be stored in the - {{{../../api/org/apache/hadoop/mapreduce/lib/output/SequenceFileOutputFormat.html} - SequenceFileOutputFormat}}, the required <<<SequenceFile.CompressionType>>> - (i.e. <<<RECORD>>> / <<<BLOCK>>> - defaults to <<<RECORD>>>) can be specified - via the SequenceFileOutputFormat.setOutputCompressionType(Job, - SequenceFile.CompressionType) api. - -*** Skipping Bad Records - - Hadoop provides an option where a certain set of bad input records can be - skipped when processing map inputs. Applications can control this feature - through the {{{../../api/org/apache/hadoop/mapred/SkipBadRecords.html} - SkipBadRecords}} class. - - This feature can be used when map tasks crash deterministically on certain - input. This usually happens due to bugs in the map function. Usually, the - user would have to fix these bugs. This is, however, not possible sometimes. - The bug may be in third party libraries, for example, for which the source - code is not available. In such cases, the task never completes successfully - even after multiple attempts, and the job fails. With this feature, only a - small portion of data surrounding the bad records is lost, which may be - acceptable for some applications (those performing statistical analysis on - very large data, for example). - - By default this feature is disabled. For enabling it, refer to - {{{../../api/org/apache/hadoop/mapred/SkipBadRecords.html} - SkipBadRecords.setMapperMaxSkipRecords(Configuration, long)}} and - {{{../../api/org/apache/hadoop/mapred/SkipBadRecords.html} - SkipBadRecords.setReducerMaxSkipGroups(Configuration, long)}}. - - With this feature enabled, the framework gets into 'skipping mode' after a - certain number of map failures. For more details, see - {{{../../api/org/apache/hadoop/mapred/SkipBadRecords.html} - SkipBadRecords.setAttemptsToStartSkipping(Configuration, int)}}. In 'skipping - mode', map tasks maintain the range of records being processed. To do this, - the framework relies on the processed record counter. See - {{{../../api/org/apache/hadoop/mapred/SkipBadRecords.html} - SkipBadRecords.COUNTER_MAP_PROCESSED_RECORDS}} and - {{{../../api/org/apache/hadoop/mapred/SkipBadRecords.html} - SkipBadRecords.COUNTER_REDUCE_PROCESSED_GROUPS}}. This counter enables the - framework to know how many records have been processed successfully, and - hence, what record range caused a task to crash. On further attempts, - this range of records is skipped. - - The number of records skipped depends on how frequently the processed record - counter is incremented by the application. It is recommended that this - counter be incremented after every record is processed. This may not be - possible in some applications that typically batch their processing. In such - cases, the framework may skip additional records surrounding the bad record. - Users can control the number of skipped records through - {{{../../api/org/apache/hadoop/mapred/SkipBadRecords.html} - SkipBadRecords.setMapperMaxSkipRecords(Configuration, long)}} and - {{{../../api/org/apache/hadoop/mapred/SkipBadRecords.html} - SkipBadRecords.setReducerMaxSkipGroups(Configuration, long)}}. The framework - tries to narrow the range of skipped records using a binary search-like - approach. The skipped range is divided into two halves and only one half gets - executed. On subsequent failures, the framework figures out which half - contains bad records. A task will be re-executed till the acceptable skipped - value is met or all task attempts are exhausted. To increase the number of - task attempts, use - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setMaxMapAttempts(int)}} and - {{{../../api/org/apache/hadoop/mapreduce/Job.html} - Job.setMaxReduceAttempts(int)}} - - Skipped records are written to HDFS in the sequence file format, for later - analysis. The location can be changed through - {{{../../api/org/apache/hadoop/mapred/SkipBadRecords.html} - SkipBadRecords.setSkipOutputPath(JobConf, Path)}}. - -** Example: WordCount v2.0 - - Here is a more complete <<<WordCount>>> which uses many of the features - provided by the MapReduce framework we discussed so far. - - This needs the HDFS to be up and running, especially for the - <<<DistributedCache>>>-related features. Hence it only works with a - {{{../../hadoop-project-dist/hadoop-common/SingleCluster.html} - pseudo-distributed}} or - {{{../../hadoop-project-dist/hadoop-common/ClusterSetup.html} - fully-distributed}} Hadoop installation. - -*** Source Code - -+---+ -import java.io.BufferedReader; -import java.io.FileReader; -import java.io.IOException; -import java.net.URI; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.List; -import java.util.Set; -import java.util.StringTokenizer; - -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.io.IntWritable; -import org.apache.hadoop.io.Text; -import org.apache.hadoop.mapreduce.Job; -import org.apache.hadoop.mapreduce.Mapper; -import org.apache.hadoop.mapreduce.Reducer; -import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; -import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; -import org.apache.hadoop.mapreduce.Counter; -import org.apache.hadoop.util.GenericOptionsParser; -import org.apache.hadoop.util.StringUtils; - -public class WordCount2 { - - public static class TokenizerMapper - extends Mapper<Object, Text, Text, IntWritable>{ - - static enum CountersEnum { INPUT_WORDS } - - private final static IntWritable one = new IntWritable(1); - private Text word = new Text(); - - private boolean caseSensitive; - private Set<String> patternsToSkip = new HashSet<String>(); - - private Configuration conf; - private BufferedReader fis; - - @Override - public void setup(Context context) throws IOException, - InterruptedException { - conf = context.getConfiguration(); - caseSensitive = conf.getBoolean("wordcount.case.sensitive", true); - if (conf.getBoolean("wordcount.skip.patterns", true)) { - URI[] patternsURIs = Job.getInstance(conf).getCacheFiles(); - for (URI patternsURI : patternsURIs) { - Path patternsPath = new Path(patternsURI.getPath()); - String patternsFileName = patternsPath.getName().toString(); - parseSkipFile(patternsFileName); - } - } - } - - private void parseSkipFile(String fileName) { - try { - fis = new BufferedReader(new FileReader(fileName)); - String pattern = null; - while ((pattern = fis.readLine()) != null) { - patternsToSkip.add(pattern); - } - } catch (IOException ioe) { - System.err.println("Caught exception while parsing the cached file '" - + StringUtils.stringifyException(ioe)); - } - } - - @Override - public void map(Object key, Text value, Context context - ) throws IOException, InterruptedException { - String line = (caseSensitive) ? - value.toString() : value.toString().toLowerCase(); - for (String pattern : patternsToSkip) { - line = line.replaceAll(pattern, ""); - } - StringTokenizer itr = new StringTokenizer(line); - while (itr.hasMoreTokens()) { - word.set(itr.nextToken()); - context.write(word, one); - Counter counter = context.getCounter(CountersEnum.class.getName(), - CountersEnum.INPUT_WORDS.toString()); - counter.increment(1); - } - } - } - - public static class IntSumReducer - extends Reducer<Text,IntWritable,Text,IntWritable> { - private IntWritable result = new IntWritable(); - - public void reduce(Text key, Iterable<IntWritable> values, - Context context - ) throws IOException, InterruptedException { - int sum = 0; - for (IntWritable val : values) { - sum += val.get(); - } - result.set(sum); - context.write(key, result); - } - } - - public static void main(String[] args) throws Exception { - Configuration conf = new Configuration(); - GenericOptionsParser optionParser = new GenericOptionsParser(conf, args); - String[] remainingArgs = optionParser.getRemainingArgs(); - if (!(remainingArgs.length != 2 || remainingArgs.length != 4)) { - System.err.println("Usage: wordcount <in> <out> [-skip skipPatternFile]"); - System.exit(2); - } - Job job = Job.getInstance(conf, "word count"); - job.setJarByClass(WordCount2.class); - job.setMapperClass(TokenizerMapper.class); - job.setCombinerClass(IntSumReducer.class); - job.setReducerClass(IntSumReducer.class); - job.setOutputKeyClass(Text.class); - job.setOutputValueClass(IntWritable.class); - - List<String> otherArgs = new ArrayList<String>(); - for (int i=0; i < remainingArgs.length; ++i) { - if ("-skip".equals(remainingArgs[i])) { - job.addCacheFile(new Path(remainingArgs[++i]).toUri()); - job.getConfiguration().setBoolean("wordcount.skip.patterns", true); - } else { - otherArgs.add(remainingArgs[i]); - } - } - FileInputFormat.addInputPath(job, new Path(otherArgs.get(0))); - FileOutputFormat.setOutputPath(job, new Path(otherArgs.get(1))); - - System.exit(job.waitForCompletion(true) ? 0 : 1); - } -} -+---+ - -*** Sample Runs - - Sample text-files as input: - - <<<$ bin/hdfs dfs -ls /user/joe/wordcount/input/>>> \ - <<</user/joe/wordcount/input/file01>>> \ - <<</user/joe/wordcount/input/file02>>> \ - \ - <<<$ bin/hdfs dfs -cat /user/joe/wordcount/input/file01>>> \ - <<<Hello World, Bye World!>>> \ - \ - <<<$ bin/hdfs dfs -cat /user/joe/wordcount/input/file02>>> \ - <<<Hello Hadoop, Goodbye to hadoop.>>> - - Run the application: - - <<<$ bin/hadoop jar wc.jar WordCount2 /user/joe/wordcount/input - /user/joe/wordcount/output>>> - - Output: - - <<<$ bin/hdfs dfs -cat /user/joe/wordcount/output/part-r-00000>>> \ - <<<Bye 1>>> \ - <<<Goodbye 1>>> \ - <<<Hadoop, 1>>> \ - <<<Hello 2>>> \ - <<<World! 1>>> \ - <<<World, 1>>> \ - <<<hadoop. 1>>> \ - <<<to 1>>> - - Notice that the inputs differ from the first version we looked at, and how - they affect the outputs. - - Now, lets plug-in a pattern-file which lists the word-patterns to be ignored, - via the <<<DistributedCache>>>. - - <<<$ bin/hdfs dfs -cat /user/joe/wordcount/patterns.txt>>> \ - <<<\\.>>> \ - <<<\\,>>> \ - <<<\\!>>> \ - <<<to>>> - - Run it again, this time with more options: - - <<<$ bin/hadoop jar wc.jar WordCount2 - -Dwordcount.case.sensitive=true /user/joe/wordcount/input - /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt>>> - - As expected, the output: - - <<<$ bin/hdfs dfs -cat /user/joe/wordcount/output/part-r-00000>>> \ - <<<Bye 1>>> \ - <<<Goodbye 1>>> \ - <<<Hadoop 1>>> \ - <<<Hello 2>>> \ - <<<World 2>>> \ - <<<hadoop 1>>> - - Run it once more, this time switch-off case-sensitivity: - - <<<$ bin/hadoop jar wc.jar WordCount2 - -Dwordcount.case.sensitive=false /user/joe/wordcount/input - /user/joe/wordcount/output -skip /user/joe/wordcount/patterns.txt>>> - - Sure enough, the output: - - <<<$ bin/hdfs dfs -cat /user/joe/wordcount/output/part-r-00000>>> \ - <<<bye 1>>> \ - <<<goodbye 1>>> \ - <<<hadoop 2>>> \ - <<<hello 2>>> \ - <<<horld 2>>> - -*** Highlights - - The second version of <<<WordCount>>> improves upon the previous one by using - some features offered by the MapReduce framework: - - * Demonstrates how applications can access configuration parameters in the - <<<setup>>> method of the <<<Mapper>>> (and <<<Reducer>>>) - implementations. - - * Demonstrates how the <<<DistributedCache>>> can be used to distribute - read-only data needed by the jobs. Here it allows the user to specify - word-patterns to skip while counting. - - * Demonstrates the utility of the <<<GenericOptionsParser>>> to handle - generic Hadoop command-line options. - - * Demonstrates how applications can use <<<Counters>>> and how they can set - application-specific status information passed to the <<<map>>> (and - <<<reduce>>>) method. - - <Java and JNI are trademarks or registered trademarks of Oracle America, - Inc. in the United States and other countries.>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8b787e2f/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/apt/MapReduce_Compatibility_Hadoop1_Hadoop2.apt.vm ---------------------------------------------------------------------- diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/apt/MapReduce_Compatibility_Hadoop1_Hadoop2.apt.vm b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/apt/MapReduce_Compatibility_Hadoop1_Hadoop2.apt.vm deleted file mode 100644 index e0fce63..0000000 --- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/apt/MapReduce_Compatibility_Hadoop1_Hadoop2.apt.vm +++ /dev/null @@ -1,114 +0,0 @@ -~~ Licensed under the Apache License, Version 2.0 (the "License"); -~~ you may not use this file except in compliance with the License. -~~ You may obtain a copy of the License at -~~ -~~ http://www.apache.org/licenses/LICENSE-2.0 -~~ -~~ Unless required by applicable law or agreed to in writing, software -~~ distributed under the License is distributed on an "AS IS" BASIS, -~~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -~~ See the License for the specific language governing permissions and -~~ limitations under the License. See accompanying LICENSE file. - - --- - Hadoop Map Reduce Next Generation-${project.version} - Backward Compatibility - --- - --- - ${maven.build.timestamp} - -Apache Hadoop MapReduce - Migrating from Apache Hadoop 1.x to Apache Hadoop 2.x - -* {Introduction} - - This document provides information for users to migrate their Apache Hadoop - MapReduce applications from Apache Hadoop 1.x to Apache Hadoop 2.x. - - In Apache Hadoop 2.x we have spun off resource management capabilities - into Apache Hadoop YARN, a general purpose, distributed application management - framework while Apache Hadoop MapReduce (aka MRv2) remains as a pure - distributed computation framework. - - In general, the previous MapReduce runtime (aka MRv1) has been reused and - no major surgery has been conducted on it. Therefore, MRv2 is able to ensure - satisfactory compatibility with MRv1 applications. However, due to some - improvements and code refactorings, a few APIs have been rendered - backward-incompatible. - - The remainder of this page will discuss the scope and the level of backward - compatibility that we support in Apache Hadoop MapReduce 2.x (MRv2). - -* {Binary Compatibility} - - First, we ensure binary compatibility to the applications that use old - <<mapred>> APIs. This means that applications which were built against MRv1 - <<mapred>> APIs can run directly on YARN without recompilation, merely by - pointing them to an Apache Hadoop 2.x cluster via configuration. - -* {Source Compatibility} - - We cannot ensure complete binary compatibility with the applications that use - <<mapreduce>> APIs, as these APIs have evolved a lot since MRv1. However, we - ensure source compatibility for <<mapreduce>> APIs that break binary - compatibility. In other words, users should recompile their applications that - use <<mapreduce>> APIs against MRv2 jars. One notable binary incompatibility - break is Counter and CounterGroup. - -* {Not Supported} - - MRAdmin has been removed in MRv2 because because <<<mradmin>>> commands - no longer exist. They have been replaced by the commands in <<<rmadmin>>>. We - neither support binary compatibility nor source compatibility for the - applications that use this class directly. - -* {Tradeoffs between MRv1 Users and Early MRv2 Adopters} - - Unfortunately, maintaining binary compatibility for MRv1 applications may lead - to binary incompatibility issues for early MRv2 adopters, in particular Hadoop - 0.23 users. For <<mapred>> APIs, we have chosen to be compatible with MRv1 - applications, which have a larger user base. For <<mapreduce>> APIs, if they - don't significantly break Hadoop 0.23 applications, we still change them to be - compatible with MRv1 applications. Below is the list of MapReduce APIs which - are incompatible with Hadoop 0.23. - -*-----------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------+ -| <<Problematic Function>> | <<Incompatibility Issue>> | -*-----------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------+ -| <<<org.apache.hadoop.util.ProgramDriver#drive>>> | Return type changes from <<<void>>> to <<<int>>> | -*-----------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------+ -| <<<org.apache.hadoop.mapred.jobcontrol.Job#getMapredJobID>>> | Return type changes from <<<String>>> to <<<JobID>>> | -*-----------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------+ -| <<<org.apache.hadoop.mapred.TaskReport#getTaskId>>> | Return type changes from <<<String>>> to <<<TaskID>>> | -*-----------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------+ -| <<<org.apache.hadoop.mapred.ClusterStatus#UNINITIALIZED_MEMORY_VALUE>>> | Data type changes from <<<long>>> to <<<int>>> | -*-----------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------+ -| <<<org.apache.hadoop.mapreduce.filecache.DistributedCache#getArchiveTimestamps>>> | Return type changes from <<<long[]>>> to <<<String[]>>> | -*-----------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------+ -| <<<org.apache.hadoop.mapreduce.filecache.DistributedCache#getFileTimestamps>>> | Return type changes from <<<long[]>>> to <<<String[]>>> | -*-----------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------+ -| <<<org.apache.hadoop.mapreduce.Job#failTask>>> | Return type changes from <<<void>>> to <<<boolean>>> | -*-----------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------+ -| <<<org.apache.hadoop.mapreduce.Job#killTask>>> | Return type changes from <<<void>>> to <<<boolean>>> | -*-----------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------+ -| <<<org.apache.hadoop.mapreduce.Job#getTaskCompletionEvents>>> | Return type changes from <<<o.a.h.mapred.TaskCompletionEvent[]>>> to <<<o.a.h.mapreduce.TaskCompletionEvent[]>>> | -*-----------------------------------------------------------------------------------+------------------------------------------------------------------------------------------------------------------+ - -* {Malicious} - - For the users who are going to try <<<hadoop-examples-1.x.x.jar>>> on YARN, - please note that <<<hadoop -jar hadoop-examples-1.x.x.jar>>> will still use - <<<hadoop-mapreduce-examples-2.x.x.jar>>>, which is installed together with - other MRv2 jars. By default Hadoop framework jars appear before the users' - jars in the classpath, such that the classes from the 2.x.x jar will still be - picked. Users should remove <<<hadoop-mapreduce-examples-2.x.x.jar>>> - from the classpath of all the nodes in a cluster. Otherwise, users need to - set <<<HADOOP_USER_CLASSPATH_FIRST=true>>> and - <<<HADOOP_CLASSPATH=...:hadoop-examples-1.x.x.jar>>> to run their target - examples jar, and add the following configuration in <<<mapred-site.xml>>> to - make the processes in YARN containers pick this jar as well. - -+---+ - <property> - <name>mapreduce.job.user.classpath.first</name> - <value>true</value> - </property> -+---+