Hadoop – 流媒体

Hadoop – 流媒体


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

使用 Python 的示例

对于 Hadoop 流,我们正在考虑字数问题。Hadoop 中的任何作业都必须有两个阶段:mapper 和 reducer。我们已经在python脚本中为mapper和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)

将 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

流媒体的工作原理

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

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

当为reducer 指定脚本时,每个reducer 任务都会将脚本作为单独的进程启动,然后初始化reducer。当 reducer 任务运行时,它将其输入键/值对转换为行并将这些行提供给进程的标准输入 (STDIN)。同时,reducer 从进程的标准输出(STDOUT)中收集面向行的输出,将每一行转换成一个键/值对,作为reducer的输出收集。默认情况下,直到第一个制表符的行的前缀是键,该行的其余部分(不包括制表符)是值。但是,这可以根据特定要求进行定制。

重要命令

Parameters 选项 描述
-input directory/file-name 必需的 映射器的输入位置。
-output directory-name 必需的 减速机的输出位置。
-mapper executable or script or JavaClassName 必需的 映射器可执行文件。
-reducer executable or script or JavaClassName 必需的 减速器可执行文件。
-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 可选的 当减少任务失败时调用的脚本。

觉得文章有用?

点个广告表达一下你的爱意吧 !😁