MRjob:一个减速器能执行两个操作吗



我正试图得出映射器生成的每个键值对的概率。

所以,比方说映射器收益率:

a, (r, 5)
a, (e, 6)
a, (w, 7)

我需要加5+6+7=18,然后找到概率5/18,6/18,7/18

因此减速器的最终输出看起来像:

a, [[r, 5, 0.278], [e, 6, 0.33], [w, 7, 0.389]]

到目前为止,我只能让reducer对该值中的所有整数求和。如何返回并将每个实例除以总数?

谢谢!

Pai的解决方案在技术上是正确的,但在实践中这会给您带来很多冲突,因为设置分区可能会带来很大的痛苦(请参阅https://groups.google.com/forum/#!主题/mrjob/aV7bNn0sJ2k)。

通过使用mrjob.step,然后创建两个减速器,可以更容易地完成此任务,例如本例:https://github.com/Yelp/mrjob/blob/master/mrjob/examples/mr_next_word_stats.py

按照你所描述的方式来做:

from mrjob.job import MRJob
import re
from mrjob.step import MRStep
from collections import defaultdict
wordRe = re.compile(r"[w]+")
class MRComplaintFrequencyCount(MRJob):
    def mapper(self, _, line):
        self.increment_counter('group','num_mapper_calls',1)
        #Issue is third column in csv
        issue = line.split(",")[3]
        for word in wordRe.findall(issue):
            #Send all map outputs to same reducer
            yield word.lower(), 1
    def reducer(self, key, values):
        self.increment_counter('group','num_reducer_calls',1)  
        wordCounts = defaultdict(int)
        total = 0         
        for value in values:
            word, count = value
            total+=count
            wordCounts[word]+=count
        for k,v in wordCounts.iteritems():
            # word, frequency, relative frequency 
            yield k, (v, float(v)/total)
    def combiner(self, key, values):
        self.increment_counter('group','num_combiner_calls',1) 
        yield None, (key, sum(values))

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

这会进行标准的单词计数,并主要在组合器中进行聚合,然后使用"None"作为公共密钥,因此每个单词都会在同一个密钥下间接发送到reducer。在reducer中,您可以获得单词总数并计算相对频率。

上面所做的应该也能工作,但这是假设单个键的所有数据都能放入内存。如果是这样,那么在Reducer中,您可以将所有值保存在内存中,然后计算总数,然后计算每个键值对的边际值。这通常被称为"条纹"方法。

然而,大多数情况下,这可能是真的,并且数据可能无法放入内存。在这种情况下,你必须找到一种方法,在实际的键值对之前发送值来计算你的总数,这样当它们可以用来计算边际并立即发出值时。

这是"反转顺序"设计模式的候选者。当您需要计算相对频率时,它很有用。基本思想是在Mapper的末端,为每个中间数据发出2个键值对,其中一个键值对将为所有值具有相同的公共键。这将用于计算总数。

示例:

For a, (r, 5) :
---------------
emit (a, r), 5
emit (a, *), 5

For a, (e, 6) :
---------------
emit (a, e), 6
emit (a, *), 6

For a, (w, 7) :
---------------
emit (a, w), 7
emit (a, *), 7

完成后,您需要一个分区器,该分区器将仅使用键中的第一个值对每个中间键值对进行分区。在上面使用"a"的示例中。

您还需要一个密钥排序顺序,它总是将带有*的密钥放在密钥的第二部分。

这样,所有中间键在键的第一部分都有"a",最终都会在同一个减速器中。此外,它们将以如下所示的方式进行排序-

emit (a, *), 5
emit (a, *), 6
emit (a, *), 7
emit (a, e), 6
emit (a, r), 5
emit (a, w), 7

在reducer中,当您迭代键值对时,如果键的第二部分中有*,您将不得不简单地累积键的值。然后,您可以使用累积值来计算所有其他键值对的边际值。

total = 0
for(value : values){
    if (key.second == *)
        total += value
    else
        emit (key.first , key.second, value, value/total)
}

这种设计模式通常被称为使用对方法的反转顺序。有关这个和其他设计模式的更多信息,我建议阅读本书中关于MapReduce设计模式的章节-http://lintool.github.com/MapReduceAlgorithms/.它用例子解释得很好。

您可以简单地计算总和,也可以将对保存在内存中,以发出您想要的概率,如下所示:

reduce (key, list<values>):
    int sum = 0;
    for (value in values) {
        sum = sum + value.frequency; //assuming you can extract two fields in each value: value.word and value.frequency
    }
    String outputValue = "[";
    for (value in values) { //iterate over the values once more
        outputValue = outputValue + "["+ value.word + ", " +value.frequency + ", "+ value.frequency/sum +"],"
    }
    outputValue = outputValue.replaceLast(",","]");
    emit (key, outputValue);

当然,这只是一个伪代码,因为我不习惯python,但我希望转换应该很容易。

相关内容

  • 没有找到相关文章

最新更新