Hadoop - 流媒体

Hadoop流媒体是Hadoop发行版附带的实用程序。此实用程序允许您使用任何可执行文件或脚本作为映射器和/或缩减器来创建和运行Map / Reduce作业。

使用Python的例子

对于Hadoop流媒体，我们正在考虑单词计数问题。Hadoop中的任何工作都必须有两个阶段：映射器和简化器。我们已经编写了映射器代码和python脚本中的reducer以在Hadoop下运行它。你也可以在Perl和Ruby中编写相同的代码。

映射器阶段代码

!/usr/bin/python
import sys
# Input takes from standard input for myline in sys.stdin:
# Remove whitespace either side myline = myline.strip()
# Break the line into words words = myline.split()
# Iterate the words list for myword in words:
# Write the results to standard output print '%s\t%s' % (myword, 1)

确保该文件具有执行权限（chmod + x / home / expert / hadoop-1.2.1 / mapper.py）。

减速器阶段代码

#!/usr/bin/python
from operator import itemgetter
import sys
current_word = ""
current_count = 0
word = ""
# Input takes from standard input for myline in sys.stdin:
# Remove whitespace either side myline = myline.strip()
# Split the input we got from mapper.py word, count = myline.split('\t', 1)
# Convert count variable to integer
   try:
      count = int(count)
except ValueError:
   # Count was not a number, so silently ignore this line continue
if current_word == word:
   current_count += count
else:
   if current_word:
      # Write result to standard output print '%s\t%s' % (current_word, current_count)
   current_count = count
   current_word = word
# Do not forget to output the last word if needed!
if current_word == word:
   print '%s\t%s' % (current_word, current_count)

将映射器和缩减器代码保存在Hadoop主目录中的mapper.py和reducer.py中。确保这些文件具有执行权限（chmod + x mapper.py和chmod + x reducer.py）。由于python是缩进敏感的，因此可以从下面的链接下载相同的代码。

WordCount程序的执行

$ $HADOOP_HOME/bin/hadoop jar contrib/streaming/hadoop-streaming-1.
2.1.jar \
   -input input_dirs \
   -output output_dir \
   -mapper <path/mapper.py \
   -reducer <path/reducer.py

其中“\”用于续行以实现清晰的可读性。

例如，

./bin/hadoop jar contrib/streaming/hadoop-streaming-1.2.1.jar -input myinput -output myoutput -mapper /home/expert/hadoop-1.2.1/mapper.py -reducer /home/expert/hadoop-1.2.1/reducer.py

流媒体如何工作

在上面的例子中，mapper和reducer都是python脚本，它们从标准输入读取输入并将输出发送到标准输出。该实用程序将创建一个Map / Reduce作业，将作业提交给适当的群集，并监视作业的进度直至完成。

当为映射器指定脚本时，每个映射器任务将在映射器初始化时作为单独的进程启动脚本。在映射器任务运行时，它将其输入转换为行，并将行输入到流程的标准输入（STDIN）。同时，映射程序从进程的标准输出（STDOUT）中收集面向行的输出，并将每行转换为一个键/值对，将其作为映射器的输出进行收集。默认情况下，直到第一个制表符的行的前缀是关键字，而行的其余部分（不包括制表符）将是该值。如果行中没有制表符，则整行被认为是关键字，值为空。但是，这可以根据需要进行定制。

为减速器指定脚本时，每个减速器任务将作为单独的进程启动脚本，然后初始化减速器。当Reducer任务运行时，它将其输入键/值对转换为行并将行输入到进程的标准输入（STDIN）。同时，减速器从过程的标准输出（STDOUT）中收集面向行的输出，将每行转换为键/值对，将其作为减速器的输出收集。默认情况下，直到第一个制表符的行的前缀是关键字，行的其余部分（不包括制表符）是值。但是，这可以根据具体要求进行定制。

重要的命令

参数	描述
-input directory/file-name	输入映射器的位置。（需要）
-output directory-name	减速器的输出位置。（需要）
-mapper executable or script or JavaClassName	Mapper可执行文件。（需要）
-reducer executable or script or JavaClassName	Reducer可执行文件。（需要）
-file file-name	使计算节点上的本地映射器，缩减器或组合器可执行文件可用。
-inputformat JavaClassName	你提供的类应该返回Text类的键/值对。如果未指定，则使用TextInputFormat作为默认值。
-outputformat JavaClassName	您提供的类应采用Text类的键/值对。如果未指定，则使用TextOutputformat作为默认值。
-partitioner JavaClassName	确定哪些减少密钥的类被发送到。
-combiner streamingCommand or JavaClassName	地图输出的组合器可执行文件。
-cmdenv name = value	将环境变量传递给流命令。
-inputreader	为了向后兼容性：指定一个记录阅读器类（而不是输入格式类）。
-verbose	详细输出。
-lazyOutput	懒散地创建输出。例如，如果输出格式基于FileOutputFormat，则仅在首次调用output.collect（或Context.write）时创建输出文件。
-numReduceTasks	指定减速器的数量。
-mapdebug	脚本在地图任务失败时调用。
-reducedebug	脚本在reduce任务失败时调用。