Python中的MRJob排序

Question

我有一个任务，需要我在python中使用mapper/reducer来完成客户数据的MapReduce。我有一个CSV文件，其中包含CustomerID、ProductID和花费的金额。第一个任务是确定每个客户的总花费，我很容易就完成了。接下来的部分要求我接受这个列表，并按总花费按降序排序。我在这里挣扎..。建议在另一个MapReduce之上使用MapReduce。以下是我的代码：

第1部分：

from mrjob.job import MRJob

class TotalAmountCust(MRJob):

    def mapper(self, _, line):
        (customerid, idno, amount) = line.split(',')
        yield customerid, float(amount)

    def reducer(self, customerid, amount):
        yield customerid, sum(amount)

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

第2部分：

from mrjob.job import MRJob
from mrjob.step import MRStep

class TotalAmountCustSort(MRJob):
    def steps(self):
        return [ MRStep(mapper = self.map_by, reducer = self.red_by),
                MRStep(mapper = self.map_sort, reducer = self.red_sort) ]
    def map_by(self, _, line):
        (customerid, idno, amount) = line.split(',')
        yield customerid.zfill(2), float(amount)
    def red_by(self, customerid, amount):
        yield customerid, '%04.02f' % sum(amount)
    def map_sort(self, customerid, total):
        yield float(total), customerid
    def red_sort(self, total, customerid):
        yield total, customerid
if __name__ == '__main__':
    TotalAmountCustSort.run()

第2部分有一个问题，根本不会给我一个结果。任何建议都会得到很好的推荐。我尝试研究MRJob.SORT_VALUES = True，但没有得到我所希望的结果。

Answer 1

我解决了，输出现在是有序的

from mrjob.job import MRJob
from mrjob.step import MRStep

class SpendByCustomerSorted(MRJob):

    MRJob.SORT_VALUES = True

    def steps(self):
        return [
            MRStep(mapper=self.mapper_get_orders,
                   reducer=self.reducer_totals_by_customer),
            MRStep(mapper=self.mapper_make_amounts_key,
                   reducer=self.reducer_output_results_for_single_reducer)
        ]
    def mapper_get_orders(self, _, line):
        (customerID, itemID, orderAmount) = line.split(',')
        yield customerID, float(orderAmount)

    def reducer_totals_by_customer(self, customerID, orders):
        yield customerID, sum(orders)

    def mapper_make_amounts_key(self, customerID, orderTotal):
        yield None, ("%07.02f"%float(orderTotal), customerID)

    def reducer_output_results(self, n, orderTotalCustomerIDs):
        for c in orderTotalCustomerIDs:
            yield c[1], c[0]

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