You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by om...@apache.org on 2011/03/04 05:03:29 UTC
svn commit: r1077323 - in
/hadoop/common/branches/branch-0.20-security-patches/src/docs/src/documentation/content/xdocs:
mapred_tutorial.xml streaming.xml
Author: omalley
Date: Fri Mar 4 04:03:29 2011
New Revision: 1077323
URL: http://svn.apache.org/viewvc?rev=1077323&view=rev
Log:
commit 2235f8e5dad4811341045e24e02c327e73da6021
Author: Hemanth Yamijala <yh...@yahoo-inc.com>
Date: Wed Mar 17 11:50:39 2010 +0530
MAPREDUCE:813 from https://issues.apache.org/jira/secure/attachment/12436672/patch-813-ydist.txt
+++ b/YAHOO-CHANGES.txt
+ MAPREDUCE-813. Updates Streaming and M/R tutorial documents.
+ (Corinne Chandel via ddas)
+
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/docs/src/documentation/content/xdocs/mapred_tutorial.xml
hadoop/common/branches/branch-0.20-security-patches/src/docs/src/documentation/content/xdocs/streaming.xml
Modified: hadoop/common/branches/branch-0.20-security-patches/src/docs/src/documentation/content/xdocs/mapred_tutorial.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/docs/src/documentation/content/xdocs/mapred_tutorial.xml?rev=1077323&r1=1077322&r2=1077323&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/docs/src/documentation/content/xdocs/mapred_tutorial.xml (original)
+++ hadoop/common/branches/branch-0.20-security-patches/src/docs/src/documentation/content/xdocs/mapred_tutorial.xml Fri Mar 4 04:03:29 2011
@@ -24,7 +24,7 @@
</header>
<body>
-
+
<section>
<title>Purpose</title>
@@ -140,7 +140,7 @@
<code><k3, v3></code> (output)
</p>
</section>
-
+
<section>
<title>Example: WordCount v1.0</title>
@@ -155,7 +155,7 @@
<section>
<title>Source Code</title>
-
+
<table>
<tr>
<th></th>
@@ -164,7 +164,7 @@
<tr>
<td>1.</td>
<td>
- <code>package org.myorg;</code>
+ <code>package org.myorg;</code>
</td>
</tr>
<tr>
@@ -208,7 +208,7 @@
<tr>
<td>9.</td>
<td>
- <code>import org.apache.hadoop.mapred.*;</code>
+ <code>import org.apache.hadoop.mapred.*;</code>
</td>
</tr>
<tr>
@@ -1057,7 +1057,7 @@
<code>-Djava.library.path=<></code> etc. If the
<code>mapred.{map|reduce}.child.java.opts</code> parameters contains the
symbol <em>@taskid@</em> it is interpolated with value of
- <code>taskid</code> of the map/reduce task.</p>
+ <code>taskid</code> of the Map/Reduce task.</p>
<p>Here is an example with multiple arguments and substitutions,
showing jvm GC logging, and start of a passwordless JVM JMX agent so that
@@ -1096,7 +1096,7 @@
</p>
<section>
- <title> Memory management</title>
+ <title> Memory Management</title>
<p>Users/admins can also specify the maximum virtual memory
of the launched child-task, and any sub-process it launches
recursively, using <code>mapred.{map|reduce}.child.ulimit</code>. Note
@@ -1370,6 +1370,8 @@
JobConf.setNumTasksToExecutePerJvm(int)</a></p>
</section>
+ <section>
+ <title>Configured Parameters</title>
<p>The following properties are localized in the job configuration
for each task's execution: </p>
<table>
@@ -1396,11 +1398,25 @@
<tr><td>mapred.work.output.dir</td><td> String </td>
<td>The task's temporary output directory</td></tr>
</table>
-
+
+ <p>
+ <strong>Note:</strong>
+ During the execution of a streaming job, the names of the "mapred" parameters are transformed.
+ The dots ( . ) become underscores ( _ ).
+ For example, mapred.job.id becomes mapred_job_id and mapred.jar becomes mapred_jar.
+ To get the values in a streaming job's mapper/reducer use the parameter names with the underscores.
+ </p>
+ </section>
+
+ <section>
+ <title>Task Logs</title>
<p>The standard output (stdout) and error (stderr) streams of the task
are read by the TaskTracker and logged to
<code>${HADOOP_LOG_DIR}/userlogs</code></p>
-
+ </section>
+
+ <section>
+ <title>Distributing Libraries</title>
<p>The <a href="#DistributedCache">DistributedCache</a> 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
@@ -1414,6 +1430,7 @@
distributed cache are documented at
<a href="native_libraries.html#Loading+native+libraries+through+DistributedCache">
native_libraries.html</a></p>
+ </section>
</section>
<section>
@@ -1702,7 +1719,7 @@
set by the Map/Reduce framework. So, just create any side-files in the
path returned by
<a href="ext:api/org/apache/hadoop/mapred/fileoutputformat/getworkoutputpath">
- FileOutputFormat.getWorkOutputPath() </a>from map/reduce
+ FileOutputFormat.getWorkOutputPath() </a>from Map/Reduce
task to take advantage of this feature.</p>
<p>The entire discussion holds true for maps of jobs with
@@ -1926,7 +1943,7 @@
<p>Once user configures that profiling is needed, she/he can use
the configuration property
<code>mapred.task.profile.{maps|reduces}</code> to set the ranges
- of map/reduce tasks to profile. The value can be set using the api
+ of Map/Reduce tasks to profile. The value can be set using the api
<a href="ext:api/org/apache/hadoop/mapred/jobconf/setprofiletaskrange">
JobConf.setProfileTaskRange(boolean,String)</a>.
By default, the specified range is <code>0-2</code>.</p>
@@ -1948,7 +1965,7 @@
<section>
<title>Debugging</title>
<p>The Map/Reduce framework provides a facility to run user-provided
- scripts for debugging. When a map/reduce task fails, a user can run
+ scripts for debugging. When a Map/Reduce 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
@@ -1981,7 +1998,7 @@
<p>The arguments to the script are the task's stdout, stderr,
syslog and jobconf files. The debug command, run on the node where
- the map/reduce task failed, is: <br/>
+ the Map/Reduce task failed, is: <br/>
<code> $script $stdout $stderr $syslog $jobconf </code> </p>
<p> Pipes programs have the c++ program name as a fifth argument
Modified: hadoop/common/branches/branch-0.20-security-patches/src/docs/src/documentation/content/xdocs/streaming.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/docs/src/documentation/content/xdocs/streaming.xml?rev=1077323&r1=1077322&r2=1077323&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/docs/src/documentation/content/xdocs/streaming.xml (original)
+++ hadoop/common/branches/branch-0.20-security-patches/src/docs/src/documentation/content/xdocs/streaming.xml Fri Mar 4 04:03:29 2011
@@ -18,20 +18,19 @@
<!DOCTYPE document PUBLIC "-//APACHE//DTD Documentation V2.0//EN"
"http://forrest.apache.org/dtd/document-v20.dtd">
-
<document>
<header>
<title>Hadoop Streaming</title>
-<meta name="http-equiv">Content-Type</meta>
-<meta name="content">text/html;</meta>
-<meta name="charset">utf-8</meta>
</header>
<body>
+
+<!-- HADOOP STREAMING -->
<section>
<title>Hadoop Streaming</title>
<p>
-Hadoop streaming is a utility that comes with the Hadoop distribution. The utility allows you to create and run Map/Reduce jobs with any executable or script as the mapper and/or the reducer. For example:
+Hadoop streaming is a utility that comes with the Hadoop distribution. The utility allows you to create and run Map/Reduce jobs with any executable or
+script as the mapper and/or the reducer. For example:
</p>
<source>
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
@@ -42,20 +41,30 @@ $HADOOP_HOME/bin/hadoop jar $HADOOP_HOM
</source>
</section>
+<!-- HOW STREAMING WORKS -->
<section>
-<title>How Does Streaming Work </title>
+<title>How Streaming Works </title>
<p>
-In the above example, both the mapper and the reducer are executables that read the input from stdin (line by line) and emit the output to stdout. The utility will create a Map/Reduce job, submit the job to an appropriate cluster, and monitor the progress of the job until it completes.
-</p><p>
- When an executable is specified for mappers, each mapper task will launch the executable as a separate process when the mapper is initialized. As the mapper task runs, it converts its inputs into lines and feed the lines to the stdin of the process. In the meantime, the mapper collects the line oriented outputs from the stdout of the process and converts each line into a key/value pair, which is collected as the output of the mapper. By default, the
+In the above example, both the mapper and the reducer are executables that read the input from stdin (line by line) and emit the output to stdout.
+The utility will create a Map/Reduce job, submit the job to an appropriate cluster, and monitor the progress of the job until it completes.
+</p>
+<p>
+ When an executable is specified for mappers, each mapper task will launch the executable as a separate process when the mapper is initialized.
+ As the mapper task runs, it converts its inputs into lines and feed the lines to the stdin of the process. In the meantime, the mapper collects the
+ line oriented outputs from the stdout of the process and converts each line into a key/value pair, which is collected as the output of the mapper. By default, the
<em>prefix of a line up to the first tab character</em> is the <strong>key</strong> and the rest of the line (excluding the tab character) will be the <strong>value</strong>.
If there is no tab character in the line, then entire line is considered as key and the value is null. However, this can be customized, as discussed later.
</p>
<p>
-When an executable is specified for reducers, each reducer task will launch the executable as a separate process then the reducer is initialized. As the reducer task runs, it converts its input key/values pairs into lines and feeds the lines to the stdin of the process. In the meantime, the reducer collects the line oriented outputs from the stdout of the process, converts each line into a key/value pair, which is collected as the output of the reducer. By default, the prefix of a line up to the first tab character is the key and the rest of the line (excluding the tab character) is the value. However, this can be customized, as discussed later.
-</p><p>
+When an executable is specified for reducers, each reducer task will launch the executable as a separate process then the reducer is initialized.
+As the reducer task runs, it converts its input key/values pairs into lines and feeds the lines to the stdin of the process. In the meantime, the reducer collects the
+line oriented outputs from the stdout of the process, converts each line into a key/value pair, which is collected as the output of the reducer. By default, the
+prefix of a line up to the first tab character is the key and the rest of the line (excluding the tab character) is the value. However, this can be customized, as discussed later.
+</p>
+<p>
This is the basis for the communication protocol between the Map/Reduce framework and the streaming mapper/reducer.
-</p><p>
+</p>
+<p>
You can supply a Java class as the mapper and/or the reducer. The above example is equivalent to:
</p>
<source>
@@ -73,8 +82,52 @@ with non-zero status are considered to b
</section>
+<!-- STREAMING OPTIONS-->
+ <section>
+<title>Streaming Command Options</title>
+<p>Streaming supports streaming command options as well as <a href="streaming.html#Generic+Command+Options">generic command options</a>.
+The general command line syntax is shown below. </p>
+<p><strong>Note:</strong> Be sure to place the generic options before the streaming options, otherwise the command will fail.
+For an example, see <a href="streaming.html#Making+Archives+Available+to+Tasks">Making Archives Available to Tasks</a>.</p>
+<source>bin/hadoop command [genericOptions] [streamingOptions]</source>
+
+<p>The Hadoop streaming command options are listed here:</p>
+<table>
+<tr><th>Parameter</th><th>Optional/Required </th><th>Description </th></tr>
+<tr><td> -input directoryname or filename</td><td> Required </td><td> Input location for mapper</td></tr>
+<tr><td> -output directoryname </td><td> Required </td><td> Output location for reducer</td></tr>
+<tr><td> -mapper executable or JavaClassName </td><td> Required </td><td> Mapper executable</td></tr>
+<tr><td> -reducer executable or JavaClassName</td><td> Required </td><td> Reducer executable</td></tr>
+<tr><td> -file filename</td><td> Optional </td><td> Make the mapper, reducer, or combiner executable available locally on the compute nodes</td></tr>
+<tr><td> -inputformat JavaClassName</td><td> Optional </td><td> Class you supply should return key/value pairs of Text class. If not specified, TextInputFormat is used as the default</td></tr>
+<tr><td> -outputformat JavaClassName</td><td> Optional </td><td> Class you supply should take key/value pairs of Text class. If not specified, TextOutputformat is used as the default</td></tr>
+<tr><td> -partitioner JavaClassName</td><td> Optional </td><td> Class that determines which reduce a key is sent to</td></tr>
+<tr><td> -combiner streamingCommand or JavaClassName</td><td> Optional </td><td> Combiner executable for map output</td></tr>
+<tr><td> -cmdenv name=value</td><td> Optional </td><td> Pass environment variable to streaming commands</td></tr>
+<tr><td> -inputreader</td><td> Optional </td><td> For backwards-compatibility: specifies a record reader class (instead of an input format class)</td></tr>
+<tr><td> -verbose</td><td> Optional </td><td> Verbose output</td></tr>
+<tr><td> -lazyOutput</td><td> Optional </td><td> Create output lazily. For example, if the output format is based on FileOutputFormat, the output file is created only on the first call to output.collect (or Context.write)</td></tr>
+<tr><td> -numReduceTasks</td><td> Optional </td><td> Specify the number of reducers</td></tr>
+<tr><td> -mapdebug </td><td> Optional </td><td> Script to call when map task fails </td></tr>
+<tr><td> -reducedebug </td><td> Optional </td><td> Script to call when reduce task fails </td></tr>
+</table>
+
+<section>
+<title>Specifying a Java Class as the Mapper/Reducer</title>
+<p>You can supply a Java class as the mapper and/or the reducer. </p>
+<source>
+$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
+ -input myInputDirs \
+ -output myOutputDir \
+ -mapper org.apache.hadoop.mapred.lib.IdentityMapper \
+ -reducer /bin/wc
+</source>
+<p>You can specify <code>stream.non.zero.exit.is.failure</code> as <code>true</code> or <code>false</code> to make a streaming task that exits with a non-zero
+status to be <code>Failure</code> or <code>Success</code> respectively. By default, streaming tasks exiting with non-zero status are considered to be failed tasks.</p>
+</section>
+
<section>
-<title>Package Files With Job Submissions</title>
+<title>Packaging Files With Job Submissions</title>
<p>
You can specify any executable as the mapper and/or the reducer. The executables do not need to pre-exist on the machines in the cluster; however, if they don't, you will need to use "-file" option to tell the framework to pack your executable files as a part of job submission. For example:
</p>
@@ -84,7 +137,7 @@ $HADOOP_HOME/bin/hadoop jar $HADOOP_HOM
-output myOutputDir \
-mapper myPythonScript.py \
-reducer /bin/wc \
- -file myPythonScript.py
+ -file myPythonScript.py
</source>
<p>
The above example specifies a user defined Python executable as the mapper. The option "-file myPythonScript.py" causes the python executable shipped to the cluster machines as a part of job submission.
@@ -104,18 +157,6 @@ $HADOOP_HOME/bin/hadoop jar $HADOOP_HOM
</section>
<section>
-<title>Streaming Options and Usage </title>
-
-<section>
-<title>Mapper-Only Jobs </title>
-<p>
-Often, you may want to process input data using a map function only. To do this, simply set mapred.reduce.tasks to zero. The Map/Reduce framework will not create any reducer tasks. Rather, the outputs of the mapper tasks will be the final output of the job.
-</p><p>
-To be backward compatible, Hadoop Streaming also supports the "-reduce NONE" option, which is equivalent to "-D mapred.reduce.tasks=0".
-</p>
-</section>
-
-<section>
<title>Specifying Other Plugins for Jobs </title>
<p>
Just as with a normal Map/Reduce job, you can specify other plugins for a streaming job:
@@ -126,53 +167,191 @@ Just as with a normal Map/Reduce job, yo
-partitioner JavaClassName
-combiner streamingCommand or JavaClassName
</source>
-<p>
-The class you supply for the input format should return key/value pairs of Text class. If you do not specify an input format class, the TextInputFormat is used as the default. Since the TextInputFormat returns keys of LongWritable class, which are actually not part of the input data, the keys will be discarded; only the values will be piped to the streaming mapper.
-</p><p>
-The class you supply for the output format is expected to take key/value pairs of Text class. If you do not specify an output format class, the TextOutputFormat is used as the default.
-</p>
+<p> The class you supply for the input format should return key/value pairs of Text class. If you do not specify an input format class, the TextInputFormat is used as the default.
+Since the TextInputFormat returns keys of LongWritable class, which are actually not part of the input data, the keys will be discarded; only the values will be piped to the streaming mapper.</p>
+<p>The class you supply for the output format is expected to take key/value pairs of Text class. If you do not specify an output format class, the TextOutputFormat is used as the default.</p>
</section>
<section>
-<title>Large files and archives in Hadoop Streaming </title>
-
-<p>
-The -files and -archives options allow you to make files and archives available to the tasks. The argument is a URI to the file or archive that you have already uploaded to HDFS. These files and archives are cached across jobs. You can retrieve the host and fs_port values from the fs.default.name config variable.
-</p>
-<p>
-Here are examples of the -files option:
-</p>
+<title>Setting Environment Variables</title>
+<p>To set an environment variable in a streaming command use: </p>
<source>
--files hdfs://host:fs_port/user/testfile.txt#testlink
+ -cmdenv EXAMPLE_DIR=/home/example/dictionaries/
</source>
+</section>
+</section>
+
+<!-- GENERIC COMMAND OPTIONS-->
+<section>
+<title>Generic Command Options</title>
+<p>Streaming supports <a href="streaming.html#Streaming+Command+Options">streaming command options</a> as well as generic command options.
+The general command line syntax is shown below. </p>
+<p><strong>Note:</strong> Be sure to place the generic options before the streaming options, otherwise the command will fail.
+For an example, see <a href="streaming.html#Making+Archives+Available+to+Tasks">Making Archives Available to Tasks</a>.</p>
+<source>bin/hadoop command [genericOptions] [streamingOptions]</source>
+
+<p>The Hadoop generic command options you can use with streaming are listed here:</p>
+<table>
+<tr><th>Parameter</th><th>Optional/Required </th><th>Description </th></tr>
+<tr><td> -conf configuration_file </td><td> Optional </td><td> Specify an application configuration file </td></tr>
+<tr><td> -D property=value </td><td> Optional </td><td> Use value for given property </td></tr>
+<tr><td> -fs host:port or local </td><td> Optional </td><td> Specify a namenode </td></tr>
+<tr><td> -jt host:port or local </td><td> Optional </td><td> Specify a job tracker </td></tr>
+<tr><td> -files </td><td> Optional </td><td> Specify comma-separated files to be copied to the Map/Reduce cluster </td></tr>
+<tr><td> -libjars </td><td> Optional </td><td> Specify comma-separated jar files to include in the classpath </td></tr>
+<tr><td> -archives </td><td> Optional </td><td> Specify comma-separated archives to be unarchived on the compute machines </td></tr>
+</table>
+
+<section>
+<title>Specifying Configuration Variables with the -D Option</title>
+<p>You can specify additional configuration variables by using "-D <property>=<value>".</p>
+
+<section>
+<title>Specifying Directories</title>
<p>
-In the above example, the part of the url after # is used as the symlink name that is created in the current working directory of tasks. So the tasks will have a symlink called testlink in the cwd that points to a local copy of testfile.txt. Multiple entries can be specified as:
+To change the local temp directory use:
</p>
<source>
--files hdfs://host:fs_port/user/testfile1.txt#testlink1 -files hdfs://host:fs_port/user/testfile2.txt#testlink2
+ -D dfs.data.dir=/tmp
</source>
-<p>
-The -archives option allows you to copy jars locally to the cwd of tasks and automatically unjar the files. For example:
+<p>
+To specify additional local temp directories use:
</p>
<source>
--archives hdfs://host:fs_port/user/testfile.jar#testlink3
+ -D mapred.local.dir=/tmp/local
+ -D mapred.system.dir=/tmp/system
+ -D mapred.temp.dir=/tmp/temp
</source>
-<p>
-In the example above, a symlink testlink3 is created in the current working directory of tasks. This symlink points to the directory that stores the unjarred contents of the uploaded jar file.
+<p><strong>Note:</strong> For more details on jobconf parameters see:
+<a href="ext:mapred-default">mapred-default.html</a></p>
+</section>
+
+<section>
+<title>Specifying Map-Only Jobs </title>
+ <p>
+Often, you may want to process input data using a map function only. To do this, simply set mapred.reduce.tasks to zero.
+The Map/Reduce framework will not create any reducer tasks. Rather, the outputs of the mapper tasks will be the final output of the job.
+ </p>
+ <source>
+ -D mapred.reduce.tasks=0
+ </source>
+ <p>
+To be backward compatible, Hadoop Streaming also supports the "-reduce NONE" option, which is equivalent to "-D mapred.reduce.tasks=0".
+ </p>
+</section>
+
+<section>
+<title>Specifying the Number of Reducers</title>
+ <p>
+To specify the number of reducers, for example two, use:
+ </p>
+ <source>
+ $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
+ -D mapred.reduce.tasks=2 \
+ -input myInputDirs \
+ -output myOutputDir \
+ -mapper org.apache.hadoop.mapred.lib.IdentityMapper \
+ -reducer /bin/wc
+ </source>
+ </section>
+
+ <section>
+<title>Customizing How Lines are Split into Key/Value Pairs</title>
+ <p>
+As noted earlier, when the Map/Reduce framework reads a line from the stdout of the mapper, it splits the line into a key/value pair.
+By default, the prefix of the line up to the first tab character is the key and the rest of the line (excluding the tab character) is the value.
+ </p>
+<p>
+However, you can customize this default. You can specify a field separator other than the tab character (the default),
+and you can specify the nth (n >= 1) character rather than the first character in a line (the default) as the separator between the key and value.
+For example:
</p>
-<p>
-Here's another example of the -archives option. Here, the input.txt file has two lines specifying the names of the two files: testlink/cache.txt and testlink/cache2.txt. "testlink" is a symlink to the archived directory, which has the files "cache.txt" and "cache2.txt".
+
+ <source>
+ $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
+ -D stream.map.output.field.separator=. \
+ -D stream.num.map.output.key.fields=4 \
+ -input myInputDirs \
+ -output myOutputDir \
+ -mapper org.apache.hadoop.mapred.lib.IdentityMapper \
+ -reducer org.apache.hadoop.mapred.lib.IdentityReducer
+ </source>
+ <p>
+In the above example, "-D stream.map.output.field.separator=." specifies "." as the field separator for the map outputs,
+and the prefix up to the fourth "." in a line will be the key and the rest of the line (excluding the fourth ".") will be the value.
+If a line has less than four "."s, then the whole line will be the key and the value will be an empty Text object (like the one created by new Text("")).
+ </p>
+ <p>
+Similarly, you can use "-D stream.reduce.output.field.separator=SEP" and "-D stream.num.reduce.output.fields=NUM" to specify
+the nth field separator in a line of the reduce outputs as the separator between the key and the value.
</p>
-<source>
-$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
- -input "/user/me/samples/cachefile/input.txt" \
- -mapper "xargs cat" \
- -reducer "cat" \
- -output "/user/me/samples/cachefile/out" \
- -archives 'hdfs://hadoop-nn1.example.com/user/me/samples/cachefile/cachedir.jar#testlink' \
+<p> Similarly, you can specify "stream.map.input.field.separator" and "stream.reduce.input.field.separator" as the input separator for Map/Reduce
+inputs. By default the separator is the tab character.</p>
+ </section>
+
+</section>
+
+<section>
+<title>Working with Large Files and Archives</title>
+ <p>
+The -files and -archives options allow you to make files and archives available to the tasks. The argument is a URI to the file or archive that you have already uploaded to HDFS.
+These files and archives are cached across jobs. You can retrieve the host and fs_port values from the fs.default.name config variable.
+ </p>
+<p><strong>Note:</strong>
+The -files and -archives options are generic options.
+Be sure to place the generic options before the command options, otherwise the command will fail.
+For an example, see <a href="streaming.html#The+-archives+Option">The -archives Option</a>.
+Also see <a href="streaming.html#Other+Supported+Options">Other Supported Options</a>.
+</p>
+
+<section>
+<title>Making Files Available to Tasks</title>
+
+ <p>
+The -files option creates a symlink in the current working directory of the tasks that points to the local copy of the file.
+</p>
+ <p>
+In this example, Hadoop automatically creates a symlink named testfile.txt in the current working directory of the tasks.
+This symlink points to the local copy of testfile.txt.
+ </p>
+ <source>
+-files hdfs://host:fs_port/user/testfile.txt
+ </source>
+ <p>
+Multiple entries can be specified like this:
+ </p>
+ <source>
+-files hdfs://host:fs_port/user/testfile1.txt,hdfs://host:fs_port/user/testfile2.txt
+ </source>
+ </section>
+
+ <section>
+<title>Making Archives Available to Tasks</title>
+ <p>
+The -archives option allows you to copy jars locally to the current working directory of tasks and automatically unjar the files.
+ </p>
+ <p>
+ In this example, Hadoop automatically creates a symlink named testfile.jar in the current working directory of tasks.
+ This symlink points to the directory that stores the unjarred contents of the uploaded jar file.
+ </p>
+<source>
+-archives hdfs://host:fs_port/user/testfile.jar
+</source>
+
+<p>
+In this example, the input.txt file has two lines specifying the names of the two files: cachedir.jar/cache.txt and cachedir.jar/cache2.txt.
+"cachedir.jar" is a symlink to the archived directory, which has the files "cache.txt" and "cache2.txt".
+</p>
+ <source>
+ $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
+ -archives 'hdfs://hadoop-nn1.example.com/user/me/samples/cachefile/cachedir.jar' \
-D mapred.map.tasks=1 \
-D mapred.reduce.tasks=1 \
- -D mapred.job.name="Experiment"
+ -D mapred.job.name="Experiment" \
+ -input "/user/me/samples/cachefile/input.txt" \
+ -output "/user/me/samples/cachefile/out" \
+ -mapper "xargs cat" \
+ -reducer "cat"
$ ls test_jar/
cache.txt cache2.txt
@@ -185,8 +364,8 @@ adding: cache2.txt(in = 37) (out= 35)(de
$ hadoop dfs -put cachedir.jar samples/cachefile
$ hadoop dfs -cat /user/me/samples/cachefile/input.txt
-testlink/cache.txt
-testlink/cache2.txt
+cachedir.jar/cache.txt
+cachedir.jar/cache2.txt
$ cat test_jar/cache.txt
This is just the cache string
@@ -201,144 +380,36 @@ Found 1 items
$ hadoop dfs -cat /user/me/samples/cachefile/out/part-00000
This is just the cache string
This is just the second cache string
-
-</source>
+ </source>
+ </section>
</section>
-
-<section>
-<title>Specifying Additional Configuration Variables for Jobs </title>
-<p>
-You can specify additional configuration variables by using "-D <n>=<v>". For example:
-</p>
-<source>
-$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
- -input myInputDirs \
- -output myOutputDir \
- -mapper org.apache.hadoop.mapred.lib.IdentityMapper\
- -reducer /bin/wc \
- -D mapred.reduce.tasks=2
-</source>
-<p>
-The -D mapred.reduce.tasks=2 in the above example specifies to use two reducers for the job.
-</p>
-<p>
-For more details on the jobconf parameters see:
-<a href="ext:mapred-default">mapred-default.html</a></p>
</section>
+
+<!-- MORE USAGE EXAMPLES-->
+ <section>
+<title>More Usage Examples</title>
<section>
-<title>Other Supported Options </title>
-<p>
-Other options you may specify for a streaming job are described here:
-</p>
-<table>
-<tr><th>Parameter</th><th>Optional/Required </th><th>Description </th></tr>
-
-<tr><td> -cmdenv name=value </td><td> Optional </td><td> Pass env var to streaming commands </td></tr>
-
-<tr><td> -inputreader JavaClassName </td><td> Optional </td><td> For backwards-compatibility: specifies a record reader class (instead of an input format class) </td></tr>
-<tr><td> -verbose </td><td> Optional </td><td> Verbose output </td></tr>
-</table>
-<p>
-Streaming support Hadoop generic command line options.
-
-Supported parameters are :
-The general command line syntax is :
-<br/> bin/hadoop command [genericOptions] [commandOptions]
-</p>
-
-<table>
-<tr><th>Parameter</th><th>Optional/Required </th><th>Description </th></tr>
-
-<tr><td> -conf configuration_file </td><td> Optional </td><td> specify an application configuration file </td></tr>
-<tr><td> -D property=value </td><td> Optional </td><td> use value for given property </td></tr>
-<tr><td> -fs host:port or local </td><td> Optional </td><td> specify a namenode </td></tr>
-<tr><td> -jt host:port or local </td><td> Optional </td><td> specify a job tracker </td></tr>
-<tr><td> -files </td><td> Optional </td><td> specify comma separated files to be copied to the map reduce cluster </td></tr>
-<tr><td> -archives </td><td> Optional </td><td> specify comma separated archives to be unarchived on the compute machines </td></tr>
-<tr><td> </td><td> Optional </td><td> </td></tr>
-<tr><td> -jt host:port or local </td><td> Optional </td><td> </td></tr>
-</table>
-
-<p>
-To change the local temp directory use:
-</p>
-<source>
- -D dfs.data.dir=/tmp
-</source>
-<p>
-To specify additional local temp directories use:
-</p>
-<source>
- -D mapred.local.dir=/tmp/local
- -D mapred.system.dir=/tmp/system
- -D mapred.temp.dir=/tmp/temp
-</source>
-<p>
-For more details on jobconf parameters see:
-<a href="ext:mapred-default">mapred-default.html</a></p>
-<p>
-To set an environment variable in a streaming command use:
-</p>
-<source>
--cmdenv EXAMPLE_DIR=/home/example/dictionaries/
-</source>
-</section>
-</section>
-
-<section>
-<title>More usage examples </title>
-
-<section>
-<title>Customizing the Way to Split Lines into Key/Value Pairs </title>
-<p>
-As noted earlier, when the Map/Reduce framework reads a line from the stdout of the mapper, it splits the line into a key/value pair. By default, the prefix of the line up to the first tab character is the key and the rest of the line (excluding the tab character) is the value.
-</p>
-<p>
-However, you can customize this default. You can specify a field separator other than the tab character (the default), and you can specify the nth (n >= 1) character rather than the first character in a line (the default) as the separator between the key and value. For example:
-</p>
-
-<source>
-$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
- -input myInputDirs \
- -output myOutputDir \
- -mapper org.apache.hadoop.mapred.lib.IdentityMapper \
- -reducer org.apache.hadoop.mapred.lib.IdentityReducer \
- -D stream.map.output.field.separator=. \
- -D stream.num.map.output.key.fields=4
-</source>
-<p>
-In the above example, "-D stream.map.output.field.separator=." specifies "." as the field separator for the map outputs, and the prefix up to the fourth "." in a line will be the key and the rest of the line (excluding the fourth ".") will be the value. If a line has less than four "."s, then the whole line will be the key and the value will be an empty Text object (like the one created by new Text("")).
-</p><p>
-Similarly, you can use "-D stream.reduce.output.field.separator=SEP" and "-D stream.num.reduce.output.fields=NUM" to specify the nth field separator in a line of the reduce outputs as the separator between the key and the value.
-</p>
-<p> Similarly, you can specify "stream.map.input.field.separator" and
-"stream.reduce.input.field.separator" as the input separator for map/reduce
-inputs. By default the separator is the tab character.</p>
-</section>
-
-
-<section>
-<title>A Useful Partitioner Class (secondary sort, the -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner option) </title>
-<p>
-Hadoop has a library class,
-<a href="ext:api/org/apache/hadoop/mapred/lib/keyfieldbasedpartitioner">KeyFieldBasedPartitioner</a>,
+<title>Hadoop Partitioner Class</title>
+ <p>
+ Hadoop has a library class,
+ <a href="ext:api/org/apache/hadoop/mapred/lib/keyfieldbasedpartitioner">KeyFieldBasedPartitioner</a>, p>
that is useful for many applications. This class allows the Map/Reduce
framework to partition the map outputs based on certain key fields, not
the whole keys. For example:
</p>
<source>
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
+ -D stream.map.output.field.separator=. \
+ -D stream.num.map.output.key.fields=4 \
+ -D map.output.key.field.separator=. \
+ -D mapred.text.key.partitioner.options=-k1,2 \
+ -D mapred.reduce.tasks=12 \
-input myInputDirs \
-output myOutputDir \
-mapper org.apache.hadoop.mapred.lib.IdentityMapper \
-reducer org.apache.hadoop.mapred.lib.IdentityReducer \
- -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner \
- -D stream.map.output.field.separator=. \
- -D stream.num.map.output.key.fields=4 \
- -D map.output.key.field.separator=. \
- -D mapred.text.key.partitioner.options=-k1,2\
- -D mapred.reduce.tasks=12
+ -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner
</source>
<p>
Here, <em>-D stream.map.output.field.separator=.</em> and <em>-D stream.num.map.output.key.fields=4</em> are as explained in previous example. The two variables are used by streaming to identify the key/value pair of mapper.
@@ -360,7 +431,6 @@ Output of map (the keys)</p><source>
11.11.4.1
11.12.1.1
11.14.2.2
-
</source>
<p>
Partition into 3 reducers (the first 2 fields are used as keys for partition)</p><source>
@@ -384,7 +454,7 @@ Sorting within each partition for the re
</source>
</section>
<section>
-<title>A Useful Comparator Class</title>
+<title>Hadoop Comparator Class</title>
<p>
Hadoop has a library class,
<a href="ext:api/org/apache/hadoop/mapred/lib/keyfieldbasedcomparator">KeyFieldBasedComparator</a>,
@@ -393,16 +463,16 @@ provided by the Unix/GNU Sort. For examp
</p>
<source>
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
- -input myInputDirs \
- -output myOutputDir \
- -mapper org.apache.hadoop.mapred.lib.IdentityMapper \
- -reducer org.apache.hadoop.mapred.lib.IdentityReducer \
-D mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator \
-D stream.map.output.field.separator=. \
-D stream.num.map.output.key.fields=4 \
-D map.output.key.field.separator=. \
- -D mapred.text.key.comparator.options=-k2,2nr\
- -D mapred.reduce.tasks=12
+ -D mapred.text.key.comparator.options=-k2,2nr \
+ -D mapred.reduce.tasks=12 \
+ -input myInputDirs \
+ -output myOutputDir \
+ -mapper org.apache.hadoop.mapred.lib.IdentityMapper \
+ -reducer org.apache.hadoop.mapred.lib.IdentityReducer
</source>
<p>
The map output keys of the above Map/Reduce job normally have four fields
@@ -434,7 +504,7 @@ Sorting output for the reducer(where sec
</section>
<section>
-<title>Working with the Hadoop Aggregate Package (the -reduce aggregate option) </title>
+<title>Hadoop Aggregate Package</title>
<p>
Hadoop has a library package called
<a href="ext:api/org/apache/hadoop/mapred/lib/aggregate/package-summary">Aggregate</a>.
@@ -449,12 +519,12 @@ To use Aggregate, simply specify "-reduc
</p>
<source>
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
+ -D mapred.reduce.tasks=12 \
-input myInputDirs \
-output myOutputDir \
-mapper myAggregatorForKeyCount.py \
-reducer aggregate \
-file myAggregatorForKeyCount.py \
- -D mapred.reduce.tasks=12
</source>
<p>
The python program myAggregatorForKeyCount.py looks like:
@@ -483,28 +553,36 @@ if __name__ == "__main__":
</section>
<section>
-<title>Field Selection ( similar to unix 'cut' command) </title>
+<title>Hadoop Field Selection Class</title>
<p>
-Hadoop has a library class, org.apache.hadoop.mapred.lib.FieldSelectionMapReduce, that effectively allows you to process text data like the unix "cut" utility. The map function defined in the class treats each input key/value pair as a list of fields. You can specify the field separator (the default is the tab character). You can select an arbitrary list of fields as the map output key, and an arbitrary list of fields as the map output value. Similarly, the reduce function defined in the class treats each input key/value pair as a list of fields. You can select an arbitrary list of fields as the reduce output key, and an arbitrary list of fields as the reduce output value. For example:
+Hadoop has a library class, org.apache.hadoop.mapred.lib.FieldSelectionMapReduce, that effectively allows you to process text data like the unix "cut" utility.
+The map function defined in the class treats each input key/value pair as a list of fields. You can specify the field separator (the default is the tab character).
+You can select an arbitrary list of fields as the map output key, and an arbitrary list of fields as the map output value.
+Similarly, the reduce function defined in the class treats each input key/value pair as a list of fields.
+You can select an arbitrary list of fields as the reduce output key, and an arbitrary list of fields as the reduce output value. For example:
</p>
<source>
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
- -input myInputDirs \
- -output myOutputDir \
- -mapper org.apache.hadoop.mapred.lib.FieldSelectionMapReduce\
- -reducer org.apache.hadoop.mapred.lib.FieldSelectionMapReduce\
- -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner \
-D map.output.key.field.separa=. \
-D mapred.text.key.partitioner.options=-k1,2 \
-D mapred.data.field.separator=. \
-D map.output.key.value.fields.spec=6,5,1-3:0- \
-D reduce.output.key.value.fields.spec=0-2:5- \
- -D mapred.reduce.tasks=12
+ -D mapred.reduce.tasks=12 \
+ -input myInputDirs \
+ -output myOutputDir \
+ -mapper org.apache.hadoop.mapred.lib.FieldSelectionMapReduce \
+ -reducer org.apache.hadoop.mapred.lib.FieldSelectionMapReduce \
+ -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner
</source>
+
+<p>
+The option "-D map.output.key.value.fields.spec=6,5,1-3:0-" specifies key/value selection for the map outputs.
+Key selection spec and value selection spec are separated by ":".
+In this case, the map output key will consist of fields 6, 5, 1, 2, and 3.
+The map output value will consist of all fields (0- means field 0 and all the subsequent fields).
+</p>
<p>
-The option "-D map.output.key.value.fields.spec=6,5,1-3:0-" specifies key/value selection for the map outputs. Key selection spec and value selection spec are separated by ":". In this case, the map output key will consist of fields 6, 5, 1, 2, and 3. The map output value will consist of all fields (0- means field 0 and all
-the subsequent fields).
-</p><p>
The option "-D reduce.output.key.value.fields.spec=0-2:5-" specifies
key/value selection for the reduce outputs. In this case, the reduce
output key will consist of fields 0, 1, 2 (corresponding to the original
@@ -514,17 +592,23 @@ from field 5 (corresponding to all the o
</section>
</section>
+
+<!-- FREQUENTLY ASKED QUESTIONS -->
<section>
<title>Frequently Asked Questions </title>
+<!-- QUESTION -->
<section>
-<title>How do I use Hadoop Streaming to run an arbitrary set of (semi-)independent tasks? </title>
+<title>How do I use Hadoop Streaming to run an arbitrary set of (semi) independent tasks? </title>
<p>
-Often you do not need the full power of Map Reduce, but only need to run multiple instances of the same program - either on different parts of the data, or on the same data, but with different parameters. You can use Hadoop Streaming to do this.
+Often you do not need the full power of Map Reduce, but only need to run multiple instances of the
+same program - either on different parts of the data, or on the same data, but with different parameters.
+You can use Hadoop Streaming to do this.
</p>
</section>
+<!-- QUESTION -->
<section>
<title>How do I process files, one per map? </title>
<p>
@@ -537,25 +621,28 @@ As an example, consider the problem of z
<li>The existing Hadoop Framework:<ul>
<li>Add these commands to your main function:
<source>
- FileOutputFormat.setCompressOutput(conf, true);
- FileOutputFormat.setOutputCompressorClass(conf, org.apache.hadoop.io.compress.GzipCodec.class);
- conf.setOutputFormat(NonSplitableTextInputFormat.class);
- conf.setNumReduceTasks(0);
+ FileOutputFormat.setCompressOutput(conf, true);
+ FileOutputFormat.setOutputCompressorClass(conf, org.apache.hadoop.io.compress.GzipCodec.class);
+ conf.setOutputFormat(NonSplitableTextInputFormat.class);
+ conf.setNumReduceTasks(0);
</source></li>
<li>Write your map function:
<source>
-
- public void map(WritableComparable key, Writable value,
- OutputCollector output,
- Reporter reporter) throws IOException {
+ public void map(WritableComparable key,
+ Writable value,
+ OutputCollector output,
+ Reporter reporter)
+ throws IOException {
output.collect((Text)value, null);
- }
+ }
</source></li>
<li>Note that the output filename will not be the same as the original filename</li>
</ul></li>
</ol>
</section>
+
+<!-- QUESTION -->
<section>
<title>How many reducers should I use? </title>
<p>
@@ -563,8 +650,14 @@ See the Hadoop Wiki for details: <a href
</p>
</section>
+
+<!-- QUESTION -->
<section>
-<title>If I set up an alias in my shell script, will that work after -mapper, i.e. say I do: alias c1='cut -f1'. Will -mapper "c1" work? </title>
+<title>If I set up an alias in my shell script, will that work after -mapper?</title>
+
+<p>
+For example, say I do: alias c1='cut -f1'. Will -mapper "c1" work?
+</p>
<p>
Using an alias will not work, but variable substitution is allowed as shown in this example:
</p>
@@ -576,11 +669,11 @@ charlie 80
dan 75
$ c2='cut -f2'; $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
+ -D mapred.job.name='Experiment'
-input /user/me/samples/student_marks
- -mapper \"$c2\" -reducer 'cat'
-output /user/me/samples/student_out
- -D mapred.job.name='Experiment'
-
+ -mapper \"$c2\" -reducer 'cat'
+
$ hadoop dfs -ls samples/student_out
Found 1 items/user/me/samples/student_out/part-00000 <r 3> 16
@@ -592,15 +685,25 @@ $ hadoop dfs -cat samples/student_out/pa
</source>
</section>
+
+<!-- QUESTION -->
<section>
-<title>Can I use UNIX pipes? For example, will -mapper "cut -f1 | sed s/foo/bar/g" work?</title>
+<title>Can I use UNIX pipes?</title>
+<p>
+For example, will -mapper "cut -f1 | sed s/foo/bar/g" work?
+</p>
<p>
Currently this does not work and gives an "java.io.IOException: Broken pipe" error. This is probably a bug that needs to be investigated.
</p>
</section>
+
+<!-- QUESTION -->
<section>
-<title>When I run a streaming job by <strong>distributing large executables</strong> (for example, 3.6G) through the -file option, I get a "No space left on device" error. What do I do? </title>
+<title>What do I do if I get the "No space left on device" error?</title>
+<p>
+For example, when I run a streaming job by distributing large executables (for example, 3.6G) through the -file option, I get a "No space left on device" error.
+</p>
<p>
The jar packaging happens in a directory pointed to by the configuration variable stream.tmpdir. The default value of stream.tmpdir is /tmp. Set the value to a directory with more space:
</p>
@@ -609,6 +712,8 @@ The jar packaging happens in a directory
</source>
</section>
+
+<!-- QUESTION -->
<section>
<title>How do I specify multiple input directories? </title>
<p>
@@ -618,6 +723,8 @@ You can specify multiple input directori
</source>
</section>
+
+<!-- QUESTION -->
<section>
<title>How do I generate output files with gzip format? </title>
<p>
@@ -625,6 +732,8 @@ Instead of plain text files, you can gen
</p>
</section>
+
+<!-- QUESTION -->
<section>
<title>How do I provide my own input/output format with streaming? </title>
<p>
@@ -632,6 +741,8 @@ At least as late as version 0.14, Hadoop
</p>
</section>
+
+<!-- QUESTION -->
<section>
<title>How do I parse XML documents using streaming? </title>
<p>
@@ -645,6 +756,8 @@ Anything found between BEGIN_STRING and
</p>
</section>
+
+<!-- QUESTION -->
<section>
<title>How do I update counters in streaming applications? </title>
<p>
@@ -654,6 +767,8 @@ should be sent to stderr to update the c
</p>
</section>
+
+<!-- QUESTION -->
<section>
<title>How do I update status in streaming applications? </title>
<p>
@@ -663,6 +778,17 @@ to stderr.
</p>
</section>
+
+<!-- QUESTION -->
+<section>
+<title>How do I get the JobConf variables in a streaming job's mapper/reducer?</title>
+<p>
+See <a href="mapred_tutorial.html#Configured+Parameters">Configured Parameters</a>.
+During the execution of a streaming job, the names of the "mapred" parameters are transformed. The dots ( . ) become underscores ( _ ).
+For example, mapred.job.id becomes mapred_job_id and mapred.jar becomes mapred_jar. In your code, use the parameter names with the underscores.
+</p>
+</section>
+
</section>
</body>
</document>