如何通过识别python Hadoop中的键来处理Mapreduce答案

【问题标题】：How to deal with Mapreduce by identifying the keys in python Hadoop如何通过识别python Hadoop中的键来处理Mapreduce
【发布时间】：2018-08-11 15:50:41
【问题描述】：

我有两个来自地图功能的关键值：NY 和 Others。所以，我的密钥的输出是：NY 1，或 Other 1。只有这两种情况。

我的地图功能：

    #!/usr/bin/env python
    import sys
    import csv
    import string

    reader = csv.reader(sys.stdin, delimiter=',')
    for entry in reader:
        if len(entry) == 22:
            registration_state=entry[16]
            print('{0}\t{1}'.format(registration_state,int(1)))

现在我需要使用 reducer 来处理地图输出。我的减少：

#!/usr/bin/env python
import sys
import string


currentkey = None
ny = 0
other = 0
# input comes from STDIN (stream data that goes to the program)
for line in sys.stdin:

    #Remove leading and trailing whitespace
    line = line.strip()

    #Get key/value 
    key, values = line.split('\t', 1)  
    values = int(values)
#If we are still on the same key...
    if key == 'NY':
        ny = ny + 1
    #Otherwise, if this is a new key...
    else:
        #If this is a new key and not the first key we've seen
        other = other + 1


#Compute/output result for the last key 
print('{0}\t{1}'.format('NY',ny))
print('{0}\t{1}'.format('Other',other))

根据这些，mapreduce 将给出两个输出结果文件，每个都包含 NY 和 Others 输出。即一个包含：NY 1248，Others 4677；另一个：NY 0，Others 1000。这是因为两个reduce split 从map 输出，所以生成了两个结果，通过组合（merge）最终输出将是结果。

但是，我想更改我的 reduce 或 map 函数，使每个 reduce 过程仅在一个键上，即一个 reduce 仅处理 NY 作为键值，而另一个在 Other 上工作。我希望得到像这样的结果：

NY 1258, Others 0; Another: NY 0, Others 5677.

如何调整我的功能以达到我期望的结果？

【问题讨论】：

你是怎么运行这个的？一个 reducer 已经只有一个 key。这就是 mapreduce 的工作原理
如果有帮助，请接受并支持答案。

标签： python hadoop mapreduce reducers

【解决方案1】：

您可能需要使用 Python 迭代器和生成器。 link 给出了一个很好的例子。我已经尝试用相同（未测试）重写您的代码

映射器：

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

import sys

def main(separator='\t'):
    reader = csv.reader(sys.stdin, delimiter=',')
    for entry in reader:
    if len(entry) == 22:
        registration_state=entry[16]
        print '%s%s%d' % (registration_state, separator, 1)

if __name__ == "__main__":
    main()

减速机：

!/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'):
    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()

【讨论】：