python - 在 Map reduce Python 中按文件和按日期计数

Question

我有一个映射器，可以给我这样的词，

apple 10 12012013
apple 20 12022013
apple 30 12042013
apple 40 12032013

其中第一个值是单词，后跟第三列中提到的该单词在当天出现的情况。

我写了一个reducer，它可以选择apple作为key，然后得到它的总数。

所以输出是这样的

apple 100

但我需要输出为，

100 apple 12012013:10 12022013:20 12032013:30 12042013:40 

知道我应该如何修改我的映射器吗？

我正在 Amazon EMR Hadoop Streaming 上运行此 MapReduce 作业。

编辑:下面的代码工作完美，但我得到的输出为:

4   apple   20130601
:1  20130602
:1  20130601
:1  20130602
:1  

有什么想法吗？

Answer 1

这应该可以做到:

>>> with open('filename') as f:
    dic = {}
    for line in f:
        name, quan, dt = line.split()
        dic.setdefault(name, []).append((dt, quan))
...         

>>> for k, v in dic.items():
    total = sum(int(x[1]) for x in v)
    print '{} {} {}'.format(total, k, ' '.join('{}:{}'.format(x, y) for x, y in v))
...     
100 apple 12012013:10 12022013:20 12042013:30 12032013:40

如果相同产品的数据总是分组在一起，那么您也可以使用itertools.groupby:

>>> from itertools import groupby
>>> with open('abc1') as f:
    for k, g in groupby(f, key=lambda x:x.split()[0]):
        data = [x.split()[1:] for x in g]
        total = sum(int(x[0]) for x in data)
        print '{} {} {}'.format(total, k, ' '.join('{}:{}'.format(y, x) for x, y in data))
...         
100 apple 12012013:10 12022013:20 12042013:30 12032013:40

更新:

如果输入来自文件流，那么您可以使用sys.stdin:

import sys
from itertools import groupby
for k, g in groupby(sys.stdin, key=lambda x:x.split()[0]):
    ...