Hadoop - 流式处理

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

使用 Python 的示例

对于 Hadoop streaming，我们正在考虑词频统计问题。Hadoop 中的任何作业都必须有两个阶段：mapper 和 reducer。我们已经用 python 脚本编写了 mapper 和 reducer 的代码，以便在 Hadoop 下运行它。也可以用 Perl 和 Ruby 编写相同的代码。

Mapper 阶段代码

!/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 阶段代码

#!/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 和 reducer 代码保存在 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

Streaming 如何工作

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

当为 mapper 指定脚本时，每个 mapper 任务在 mapper 初始化时都会启动该脚本作为一个单独的进程。当 mapper 任务运行时，它会将其输入转换为行并将这些行馈送到进程的标准输入 (STDIN)。同时，mapper 从进程的标准输出 (STDOUT) 收集面向行的输出，并将每一行转换为键/值对，这些键/值对作为 mapper 的输出收集。默认情况下，一行中第一个制表符字符之前的部分是键，其余部分（不包括制表符字符）是值。如果行中没有制表符字符，则整行都被视为键，而值为 null。但是，可以根据需要自定义此设置。

当为 reducer 指定脚本时，每个 reducer 任务都会启动该脚本作为一个单独的进程，然后 reducer 初始化。当 reducer 任务运行时，它会将其输入键/值对转换为行并将这些行馈送到进程的标准输入 (STDIN)。同时，reducer 从进程的标准输出 (STDOUT) 收集面向行的输出，并将每一行转换为键/值对，这些键/值对作为 reducer 的输出收集。默认情况下，一行中第一个制表符字符之前的部分是键，其余部分（不包括制表符字符）是值。但是，可以根据具体需求自定义此设置。

重要命令