Python中经典Hadoop单词计数示例中的数据流

Question

我试图理解Python中的Hadoop字数示例http://www.michael-noll.com/tutorials/writing-an-hadoop-mapreduce-program-in-python/

作者从简单版本的映射器和缩减器开始。这是缩减器(为了简洁，我删除了一些注释)

#!/usr/bin/env python

from operator import itemgetter
import sys

current_word = None
current_count = 0
word = None

# input comes from STDIN
for line in sys.stdin:
    line = line.strip()

    word, count = line.split('\t', 1)

    try:
        count = int(count)
    except ValueError:
        continue

    if current_word == word:
        current_count += count
    else:
        if current_word:
            # write result to STDOUT
            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)

作者使用以下方法测试程序：

echo "foo foo quux labs foo bar quux" | /home/hduser/mapper.py | sort -k1,1 | /home/hduser/reducer.py

因此，reducer的编写方式就好像reducer作业的输入数据是这样的:

aa 1
aa 1
bb 1
cc 1
cc 1
cc 1

我对化简器的初步理解是，给定化简器的输入数据将包含一个唯一键。因此，在前面的示例中，将需要 3 个化简器作业。我的理解不正确吗？

然后，作者提出了映射器和还原器的改进版本。这是减速器：

#!/usr/bin/env python
"""A more advanced Reducer, using Python iterators and generators."""

from itertools import groupby
from operator import itemgetter
import sys

def read_mapper_output(file, separator='\t'):
    for line in file:
        yield line.rstrip().split(separator, 1)

def main(separator='\t'):
    # input comes from STDIN (standard input)
    data = read_mapper_output(sys.stdin, separator=separator)

    for current_word, group in groupby(data, itemgetter(0)):
        try:
            total_count = sum(int(count) for current_word, count in group)
            print "%s%s%d" % (current_word, separator, total_count)
        except ValueError:
            # count was not a number, so silently discard this item
            pass

if __name__ == "__main__":
    main()

作者补充了以下警告：

注意：以下映射和化简脚本只有在 Hadoop 上下文中运行时才能“正确”工作，即在 MapReduce 作业中作为映射器和化简器运行。这意味着运行朴素的测试命令“猫 DATA | ./mapper.py |sort -k1，1 | ./reducer.py“ 将不再正常工作，因为某些功能被有意外包给 Hadoop。

我不明白为什么朴素的测试命令不适用于新版本。我认为使用 -k1，1 排序将为两个版本的化简器生成相同的输入。我错过了什么？

Answer 1

关于你的第一个问题：“我对化简器的初步理解是，给定化简器的输入数据将包含一个唯一的键。因此，在前面的示例中，将需要 3 个化简器作业。我的理解不正确吗？

MapReduce抽象和Hadoop实现该抽象之间存在差异。在抽象中，还原器与唯一键相关联。另一方面，Hadoop实现将多个键分配给同一个reducer（以避免关闭流程和启动新流程的成本）。特别是，在Hadoop流中，reducer接收与特定数量的键（可以是零、一个或多个键）相对应的键值对，但您可以保证与某个键相关联的键值将连续出现。

例如，让我们以输入“foo foo quux labs foo bar quux”的字数为例。那么可能是一个减速器接收输入“bar 1\nfoo 1\nfoo 1\nfo1”，另一个减速器接收“labs 1\nquux 1\nquux 1”。实际运行的减速器进程的数量由您使用选项决定mapred.reduce.tasks.例如，使用2个减速器，您可以这样做

 $ bin/hadoop jar contrib/streaming/hadoop-*streaming*.jar -D mapred.reduce.tasks=2 -mapper ....

关于你的第二个问题，我同意你的观点，sort-k1,1可以解决这个问题，所以我也看不出有什么问题。