MRJob -迭代值

Question

输入(名称；日期；已用)：

Alice;01/01/2020;100
Alice;02/01/2020;30
Alice;24/01/2020;50
Bob;24/01/2020;1500
Bob;24/01/2020;12
Bob;25/01/2020;16
Bob;25/01/2020;83
Bob;25/01/2020;91
Alice;13/02/2020;10
Alice;25/02/2020;3

输出必须是在至少5天内购买的人的名字。所以对于这个输入只有Alice，因为Bob在2天内买了5次。

当我试图计算这些值时，我的问题就出现了。我已经用集合解决了这个问题：

from mrjob.job import MRJob

class MR_ex2(MRJob):
    def mapper(self, _, line):
        person, day, spent_money = line.split(';')
        yield person, day

    def reducer(self, key, counts):
        counts = set(counts)
        if len(counts) >= 5:
            yield key, key

if __name__ == '__main__':
    MR_ex2.run()

但是当我尝试添加组合器时，它崩溃了，我认为这不是最好的方法。所以我试着搜索例子，但我找不到很多例子，那些没有MRJob的例子是用一些变量来保存子问题的状态，并迭代子问题的值(在这个问题中是爱丽丝和鲍勃)来连接它们，但我不知道如何在MRJob中使用这种方式。

因此，简而言之，我的问题是如何才是解决这个问题的正确方法？连接reducer中key的所有不同值，然后检查它是否为5或更大？

Answer 1

可能不是最好的答案，但如果有人来到这里，并有类似的问题。我的部分解决方法是从生成器中展开元素，然后在生成器创建set之后，这样我就避免了错误，尽管我不确定在MapReduce中使用set是否是一种好的技术。

但是代码应该是：

from mrjob.job import MRJob

class MR_ex2(MRJob):

    def mapper(self, _, line):
        persona, dia, dinero_gastado = line.split(';')
        yield persona, dia

    def combiner(self, key, values):
        counts = set([item for sublist in values for item in sublist])
        if len(counts) >= 5:
            yield key, key
        yield key, tuple(values)

    def reducer(self, key, counts):

        counts = set([item for sublist in counts for item in sublist])
        if len(counts) >= 5:
            yield key, key

if __name__ == '__main__':
    MR_ex2.run()