MapReduce：如何跟踪映射器中多行的状态（比如计算三元组）？答案

【问题标题】：MapReduce: How to keep track of states across multiple lines in the mapper (say for counting trigrams)?MapReduce：如何跟踪映射器中多行的状态（比如计算三元组）？
【发布时间】：2014-03-03 03:36:38
【问题描述】：

我正在尝试使用 Python 中的 mrjob 框架编写一个用于计算 Trigrams 的 MapReduce 程序。到目前为止，这就是我所拥有的：

from mrjob.job import MRJob

class MRTrigram(MRJob):

    def mapper(self, _, line):
        w = line.split()
        for idx,word in enumerate(w):
            if idx < len(w) - 2:
                # Generate a trigram using the current word and next 2 words
                trigram = w[idx] + " " + w[idx + 1] + " " + w[idx + 2]
                yield trigram, 1

    def reducer(self, key, values):
        yield sum(values), key

# ignore this part - its just standard bolierplate for mrjob!
if __name__ == '__main__':
    MRTrigram.run()

可以看出，我没有处理过三元组跨行拆分的情况（例如，“它是”在第 3 行的末尾，“最好的时代”在第 4 行的开头 - 但是在这种情况下，我的代码不会捕获三元组“它是”！）。

如何跨多个映射调用保留状态，以确保无论映射器由底层运行时分配作业，只计算连续行的三元组？我曾想过将每行的最后 2 个单词存储在 MRTrigram 类中的持久数据结构中，但后来我意识到我无法保证我是否在比较第 i 行和第 i+1 行（而不是第 i、j 行，其中j 可以是文档中的任何位置！）。

有什么想法可以让我走上正轨吗？

【问题讨论】：

真的，没有人，在星期一，关于 MWE 的问题和明确的查询？也许它不像我想象的那么简单！ :)
我也对此感兴趣。每个在线实现似乎都忽略了这个问题，下面的答案没有帮助。

标签： python mapreduce mrjob

【解决方案1】：

writing a custom protocol 可能会提示您如何做到这一点，但我相信 mrjob 在您可以添加自定义行为（即形成键和值）之前，会采用由换行符分隔的流输入，所以使用 mrjob 可能无法实现。

如果您使用的是 Hadoop（即原生 Java），那么您可以编写一个自定义输入格式，该格式采用多行文本并从中解析出一个键值对。

【讨论】：