如何使用 MapReduce 对键值对进行排序？

Question

我一直在搞乱 MapReduce，对它还是很陌生，想知道是否可以就我无法回答的问题获得一些帮助： 我有一个日期和计数的 txt 文件，并希望根据它们各自的计数按升序对日期进行排序。文本文件如下所示：

我环顾四周，发现了一些这样的代码：

import re

从 mrjob.job 导入 MRJob 从 mrjob.step 导入 MRStep

WORD_RE = re.compile(r"[\w']+")

类 MRWordFrequencyCount(MRJob)：

def steps(self):
    return [
        MRStep(
            mapper=self.mapper_extract_words, combiner=self.combine_word_counts,
            reducer=self.reducer_sum_word_counts
        ),
        MRStep(
            reducer=self.reduce_sort_counts
        )
    ]

def mapper_extract_words(self, _, line):
    for word in WORD_RE.findall(line):
        yield word.lower(), 1

def combine_word_counts(self, word, counts):
    yield word, sum(counts)

def reducer_sum_word_counts(self, key, values):
    yield None, (sum(values), key)

def reduce_sort_counts(self, _, word_counts):
    for count, key in sorted(word_counts, reverse=True):
        yield ('%020d' % int(count), key)

但这似乎太复杂了，因为正如您从postedDates txt 文件中看到的那样，我已经有了密钥及其各自的计数。那么我是否只需要添加第二步，它只是一个使用“sorted(counts)”对键和值列表进行排序的 reducer 函数？

恭喜您的时间。

Answer 1

您是对的，鉴于您的特定设置，当然可以使用单个 MapReduce 执行您的任务。

您可以跳过示例中的初始步骤，因为您已经有了每个日期（键）的计数。您只需执行第二步，将这些对组合成元组并根据count 和date 进行排序

---

按升序计数

import datetime

from mrjob.job import MRJob


class MRDateFrequencyCount(MRJob):

    def mapper(self, _, line):
        date, count = line.split(' ')
        yield None, (int(count), date)

    def reducer(self, _, dates):
        for count, date in sorted(dates, key=lambda x: (x[0], datetime.datetime.strptime(x[1], '"%Y-%m-%d"'))):
            yield date, count


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

产生输出：

"\"2006-11-01\""    1
"\"2006-12-21\""    1
"\"2006-12-11\""    2
"\"2007-03-12\""    3

---

按降序计数

import datetime

from mrjob.job import MRJob


class MRDateFrequencyCount(MRJob):

    def mapper(self, _, line):
        date, count = line.split(' ')
        yield None, (int(count), date)

    def reducer(self, _, dates):
        for count, date in sorted(dates, key=lambda x: (-x[0], datetime.datetime.strptime(x[1], '"%Y-%m-%d"'))):
            yield date, count


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

产生输出：

"\"2007-03-12\""    3
"\"2006-12-11\""    2
"\"2006-11-01\""    1
"\"2006-12-21\""    1

注意：如果您图像中的数据格式与我在下面测试的文本中显示的格式不同，您将需要更改strptime 格式字符串'"%Y-%m-%d"'。

---

两个 MRJobs 都在没有配置的情况下运行，并且在包含以下文本的文本文档上运行：

"2006-12-21" 1
"2007-03-12" 3
"2006-11-01" 1
"2006-12-11" 2

当然，如果您想更改哪一列（计数或日期）排在第一位或第二位，您可以在任一减速器中更改 yeild。

您还可以使用字符串格式来删除数据集中日期周围的"。