Hadoop - 数据流
Hadoop 数据流是 Hadoop 发行版附带的实用程序。 此实用程序允许您使用任何可执行文件或脚本作为映射器或减速器来创建和运行 Map/Reduce 作业。
使用 Python 的示例
对于 Hadoop 流,我们正在考虑字数问题。 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
数据流的工作原理
在上面的示例中,mapper 和 reducer 都是 python 脚本,它们从标准输入读取输入并将输出发送到标准输出。 该实用程序将创建一个 Map/Reduce 作业,将作业提交到适当的集群,并监控作业的进度,直到它完成。
当为 mapper 指定脚本时,每个 reducer 任务将在 reducer 初始化时作为单独的进程启动脚本。 当 reducer 任务运行时,它将其输入转换为行并将这些行提供给进程的标准输入 (STDIN)。 同时,reducer 从进程的标准输出(STDOUT)中收集面向行的输出,并将每一行转换为键/值对,作为映射器的输出。 默认情况下,到第一个制表符的行的前缀是键,该行的其余部分(不包括制表符)将是值。 如果该行中没有制表符,则将整行视为键,值为空。 但是,这可以根据需要进行定制。
当为 reducer 指定脚本时,每个 reducer 任务都会将脚本作为单独的进程启动,然后初始化 reducer。 当 reducer 任务运行时,它将其输入键/值对转换为行并将这些行提供给进程的标准输入 (STDIN)。 同时,reducer 从进程的标准输出(STDOUT)中收集面向行的输出,将每一行转换成一个键/值对,作为reducer 的输出收集。 默认情况下,到第一个制表符的行的前缀是键,行的其余部分(不包括制表符)是值。 但是,这可以根据特定要求进行定制。
重要命令
参数 | 选项 | 描述 |
---|---|---|
-input directory/file-name | 必填 | mapper 的输入位置。 |
-output directory-name | 必填 | reducer 的输出位置。 |
-mapper executable or script or JavaClassName | 必填 | Mapper 可执行文件。 |
-reducer executable or script or JavaClassName | 必填 | Reducer 可执行文件。 |
-file file-name | 可选 | 使 mapper、reducer 或 combiner 可执行文件在计算节点上本地可用。 |
-inputformat JavaClassName | 可选 | 您提供的类应该返回 Text 类的键/值对。 如果未指定,则默认使用 TextInputFormat。 |
-outputformat JavaClassName | 可选 | 您提供的类应该采用 Text 类的键/值对。 如果未指定,则使用 TextOutputformat 作为默认格式。 |
-partitioner JavaClassName | 可选 | 确定将 key 发送到哪个 reduce 的类。 |
-combiner streamingCommand or JavaClassName | 可选 | 地图输出的 Combiner 可执行文件。 |
-cmdenv name=value | 可选 | 将环境变量传递给流式命令。 |
-inputreader | 可选 | 为了向后兼容:指定一个记录读取器类(而不是输入格式类)。 |
-verbose | 可选 | 详细输出。 |
-lazyOutput | 可选 | 延迟创建输出。 例如,如果输出格式基于 FileOutputFormat,则仅在第一次调用 output.collect(或 Context.write)时创建输出文件。 |
-numReduceTasks | 可选 | 指定 reducer 的数量。 |
-mapdebug | 可选 | map 任务失败时调用的脚本。 |
-reducedebug | 可选 | reduce 任务失败时调用的脚本。 |