Hadoop-Streaming(流)

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

原文链接：http://blogxinxiucan.sh1.newtouch.com/2017/07/17/Hadoop-Streaming-流/

使用Python的例子

对于Hadoop流，我们正在考虑字数问题。Hadoop中的任何工作必须有两个阶段：mapper和reducer。我们为python脚本中的mapper和reducer编写了代码，以便在Hadoop下运行它。也可以在Perl和Ruby中写同一个。

Mapper Phase代码

!/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）。

Reducer Phase代码

#!/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)

将mapper.py和reducer.py中的mapper和reducer代码保存在Hadoop主目录中。确保这些文件具有执行权限（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指定脚本时，每个reducer任务将作为单独的进程启动脚本，然后初始化reducer。当reducer任务运行时，它将其输入的键/值对转换为行，并将行馈送到进程的标准输入（STDIN）。同时，减速器从进程的标准输出（STDOUT）收集线性输出，将每行转换为键/值对，并将其作为reducer的输出进行收集。默认情况下，直到第一个制表符字符的行的前缀是键，而该行的其余部分（不包括制表符）是该值。但是，这可以根据具体要求进行定制。

重要命令