如何让 Reducer 根据键类型发出答案

【问题标题】：How to get the Reducer to emit according to key type如何让 Reducer 根据键类型发出
【发布时间】：2016-03-20 09:48:05
【问题描述】：

作为this 问题的后续，我有一个 Mapper，它正在处理大量数据并将 ID 号作为值为 1 的键发出。每个键都有两个部分，由管道分隔符分隔，用于示例：

我想要做的是让Reducer 解析键，并且对于'a' 类型的每个键，即'a|abc'，我希望Reducer 只发出重复项，但对于其他所有类型（例如键入'b'，即'b|abc'），我希望Reducer 发出所有内容，即使该值只有1。

所以上面的数据会产生：
a|abc 2
b|efg 1
b|dhh 2

在这种情况下，不会发出“a|cba 1”，因为它是“a”类型的键并且没有重复项。下面是我尝试过的代码，它几乎按预期工作，除了我得到 92 个额外的发射，其中密钥类型为“a”且计数为 1。注意：根据我的 MapReduce 日志，92 是 Reduce 任务的数量.

由于我只想要键类型“a”的重复项，我该如何修复 Reducer，以便我不会得到值 1 的键类型“a”的额外 92 次发射？

import sys
import codecs

sys.stdout = codecs.getwriter('utf-8')(sys.stdout)
inData = codecs.getreader('utf-8')(sys.stdin)

(last_key, tot_cnt) = (None, 0)
for line in inData:
    (key, val) = line.strip().split("\t")
    if last_key != key:
        k = key.split('|')
        v_id = k[0]
        if v_id == 'a':
            if tot_cnt > 1:
                sys.stdout.write("%s\t%s\n" % (last_key,tot_cnt))
        else:
            sys.stdout.write("%s\t%s\n" % (last_key,tot_cnt))

        (last_key, tot_cnt) = (key, int(val))
    else:
        (last_key, tot_cnt) = (key, tot_cnt + int(val))
if last_key:
    if v_id == 'a':
        if tot_cnt > 1:
            sys.stdout.write("%s\t%s\n" % (last_key, tot_cnt))
    else:
        sys.stdout.write("%s\t%s\n" % (last_key, tot_cnt))

【问题讨论】：

我想，我知道问题出在哪里。我会发布答案
克里斯尼尔森，我已经添加了答案。检查并确认它是否有效。

标签： python hadoop mapreduce reduce

【解决方案1】：

以下是代码中的错误：

在全局级别声明v_id，使其在任何地方都可见。

改变这一行：
```
(last_key, tot_cnt) = (None, 0)
```
收件人：
```
(last_key, tot_cnt, v_id) = (None, 0, None)
```

以下拆分应在last_key 上，而不是当前key。当前键为“b|dhh”且最后一个键为“a|abc”时，您应该为“a|abc”获得v_id。

更改此代码：

if last_key != key:
    k = key.split('|')
    v_id = k[0]
if v_id == 'a':
    if tot_cnt > 1:
        sys.stdout.write("%s\t%s\n" % (last_key,tot_cnt))
else:
    sys.stdout.write("%s\t%s\n" % (last_key,tot_cnt))

收件人：

if last_key != key:
    if last_key != None:
        k = last_key.split('|')
        v_id = k[0]

        if v_id == 'a':
            if tot_cnt > 1:
                sys.stdout.write("%s\t%s\n" % (last_key,tot_cnt))
        else:
            sys.stdout.write("%s\t%s\n" % (last_key,tot_cnt))

所以，修改后的 reducer 代码如下所示：

import sys
import codecs

sys.stdout = codecs.getwriter('utf-8')(sys.stdout)
inData = codecs.getreader('utf-8')(sys.stdin)

(last_key, tot_cnt, v_id) = (None, 0, None)

for line in inData:
    (key, val) = line.strip().split("\t")
    if last_key != key:
        if last_key != None:
            k = last_key.split('|')
            v_id = k[0]

            if v_id == 'a':
                if tot_cnt > 1:
                    sys.stdout.write("%s\t%s\n" % (last_key,tot_cnt))
            else:
                sys.stdout.write("%s\t%s\n" % (last_key,tot_cnt))

        (last_key, tot_cnt) = (key, int(val))
    else:
        (last_key, tot_cnt) = (key, tot_cnt + int(val))

if last_key:
    if v_id == 'a':
        if tot_cnt > 1:
            sys.stdout.write("%s\t%s\n" % (last_key, tot_cnt)) 
    else:
        sys.stdout.write("%s\t%s\n" % (last_key, tot_cnt))

当我运行这个时，我得到了输出：

a|abc   2
b|dhh   2
b|efg   1

注意：我不是 Python 专家。我觉得，你可以优化这段代码。因此，请检查脚本中是否存在任何极端情况和冗余检查。

【讨论】：