首页 > 代码库 > Hadoop-2.4.1学习之Streaming编程
Hadoop-2.4.1学习之Streaming编程
在之前的文章曾提到Hadoop不仅支持用Java编写的job，也支持其它语言编写的作业，比如Hadoop Streaming（shell、python）和Hadoop Pipes（c++），本篇文章将学习Hadoop Streaming编程。Streaming是hadoop自带的工具，封装在hadoop-streaming-版本号.jar中，可以使用hadoop jar hadoop-streaming-版本号.jar命令启动，在该命令中还需要指定mapper或/和reducer，其中mapper和reducer任务既可以是java类，也可以是可执行文件（如cat）或脚本文件（如python）。该工具将创建MapReduce作业，将作业提交给集群处理并监控作业的执行进度等，下面分析一下Streaming的执行过程。
当可执行文件做为mapper或者reducer时，每个mapper任务或者reducer任务在mapper或者reducer初始化后将可执行文件做为独立的进程加载运行。当mapper任务运行时，它将输入转换为行然后把这些行传递给该进程的标准输入stdin，同时mapper收集从该进程的标准输出stdout产生的行并将每行转换为键值对，这些键值对将做为该mapper的输出。默认的情况下，一行的开始部分到第一个tab符为键，剩余的部分（不包括tab符）为值，如果某行不存在tab符，则整个行将做为键，而值为null。但可通过使用-inputformat命令行选项自定义键值分隔符。当reducer任务运行时，它将输入的键值对转换为行，然后传递给该进程的标准输入stdin，同时reducer任务收集该进程标准输出stdout的输出，并转换为键值对，这些键值对将做为reducer任务的输出。同mapper任务一样，默认使用tab符分隔键值对，用户可以使用-outputformat自定义分隔符。
通过上面的描述可知，mapper和reducer的的输出都为键值对，输入都为从相应进程的stdin输入的行，不同的是mapper是将输入数据转换为行，reducer是将mapper输出的键值对转换为行。
可以通过指定stream.non.zero.exit.is.failure的值为true或false表示streaming任务退出时状态码的含义，如果为true则非0值表示失败，若为false则非0表示成功，默认情况下，streaming任务退出时的状态码非0表示任务失败。接下来看看streaming工具的语法格式，用户可以通过在命令行输入hadoop jar hadoop-streaming-2.4.1.jar –help获取该工具的具体用法，如下表所示：
参数 | 可选/必选 | 说明 |
-input | 必选 | Map阶段输入文件的路径 |
-output | 必选 | Reduce阶段的输出目录 |
-mapper | 可选 | 可执行文件或者Java类，做为mapper。默认为PipeMapper |
-reducer | 可选 | 可执行文件或者Java类，做为reducer。默认为PipeReducer |
-combiner | 可选 | 可执行文件或者Java类，做为combiner。默认为PipeCombiner |
-partitioner | 可选 | 做为partitioner的Java类 |
-inputformat | 可选 | 指定输入格式的Java类，默认为TextInputFormat，还可以为SequenceFileAsTextInputFormat或者自定义输入格式的Java类 |
-outputformat | 可选 | 指定输出格式的Java类，默认为TextOutputFormat，也可以为自定义输出格式的Java类 |
-file | 可选 | 指定了作业使用的文件，将被拷贝到集群中。推荐使用通用选项-files选线替代该选项 |
-numnumReduceTasks | 可选 | 指定reducer的数量 |
-inputreader | 可选 | 指定读取记录的reader类 |
-cmdenv | 可选 | <n>=<v>，像streaming传递环境变量 |
-mapdebug | 可选 | 指定了map任务失败时运行的脚本 |
-reducedebug | 可选 | 指定了reduce任务失败时运行的脚本 |
-lazyOutput | 可选 | 延迟创建输出，例如如果输出为TextOutputFormat，输出文件仅在第一次调用Context.write时创建 |
-background | 可选 | 提交作业后立即返回，不等待作业完成 |
-verbose | 可选 | 打印作业的执行情况 |
-info | 可选 | 打印详细的使用方式 |
除了上表所述的专门用于streaming的选项外，在使用streaming工具时还可以指定通用选项，但需要确保通用选项位于streaming选项之前，否则将导致失败。通用选项如下表所示：
参数 | 可选/必选 | 说明 |
-conf configuration_file | 可选 | 指定应用程序的配置文件 |
-D property=value | 可选 | 为指定属性设置特定值 |
-fs host:port or local | 可选 | 指定NameNode |
-files | 可选 | 指定用逗号分隔的传递到MapReduce集群的文件 |
-libjars | 可选 | 指定要被包含在类路径中的用逗号分隔的jar文件 |
-archives | 可选 | 指定了用逗号分隔的归档文件，这些归档文件将在计算节点上解压缩 |
前面曾经提到，默认使用tab符分隔键值对，并且按照第一个tab符来分隔，而在很多情况下，数据不是使用tab符分隔字段，并且希望某几个字段做为键，默认情况将不满足这样的需求，此时用户可以通过使用相应的参数来修改默认设置，这几个参数为：
参数 | 说明 |
stream.map.input.field.separator | Map输入的字段分隔符，默认为\t |
stream.map.output.field.separator | Map输出的字段分隔符，默认为\t |
stream.num.map.output.key.fields | 第几个分隔符用于分隔键值对，默认为1 |
stream.reduce.input.field.separator | Reduce输入的字段分隔符，默认为\t |
stream.reduce.output.field.separator | Reduce输出的字段分隔符，默认为\t |
stream.num.reduce.output.key.fields | 第几个分隔符用于分隔键值对，默认为1 |
通过一个具体的示例代码来详细描述上述参数的含义，在该段代码中使用点号（.）做为分隔符，并且第四个点号之前的字段为键，第四个点号（不包含该点号）后面的字段做为值，如果某行中的点号少于四个，则正行将做为键，值为空的Text对象。代码如下：
hadoop jar hadoop-streaming-2.4.1.jar -D stream.map.output.field.separator=. -D stream.num.map.output.key.fields=4 -input input -output output -mapper /bin/cat -reducer /bin/cat
最后通过下面的示例代码结束Streaming编程的学习，在该代码中通过-files通用选项将两个python脚本上传到集群中，并分别做为mapper和reducer：
hadoop jar hadoop-streaming-2.4.1.jar -files mapperPythonScript.py, reducerPythonScript.py -input myInputDirs -output myOutputDir -mapper mapperPythonScript.py -reducer reducerPythonScript.py
本篇文章学习Hadoop Streaming编程，详细介绍了作业流程和参数的使用方式，至于如何编写Streaming中的mapper和reducer，则需要根据用户使用的脚本语言（如python、shell）而定。
Hadoop-2.4.1学习之Streaming编程