Проблема при использовании SORT_VALUES в задании MapReduce с использованием mrjob (значения ключей не сортируются на входе редуктора) - PullRequest
Новая
       33

Проблема при использовании SORT_VALUES в задании MapReduce с использованием mrjob (значения ключей не сортируются на входе редуктора)

3 голосов
/ 21 мая 2019

Я хочу создать программу MapReduce, редуктор которой получает пары k-v, отсортированные по значению. Я использую mrjob, чей параметр SORT_VALUES, казалось, идеально подходит для этой задачи. После установки этого параметра в True вход редуктора не сортируется, например, я получаю следующее (учитывая, что A должен предшествовать X): 

"ES"    ["X", 3]
"ES"    ["A", "Spain"]

Я использую Python 2.7.5, mrjob == 0.6.1 и Hadoop. Локальное выполнение программы дает мне: 

"ES"    ["A", "Spain"]
"ES"    ["X", 1]
"ES"    ["X", 2]

Что правильно. Но выполнение Hadoop дает: 

"ES"    ["X", 3]
"ES"    ["A", "Spain"]

Мой код: 


import sys, os, re
from mrjob.job import MRJob
from mrjob.step import MRStep 

class MRJoin(MRJob):

    SORT_VALUES = True

    def mapper(self, _, line):
        splits = line.rstrip("\n").split(",")
        if len(splits) == 2: # countries
            symbol = 'A' # countries before clients
            country2digit = splits[1]
            yield country2digit, (symbol, splits[0])
        else: #  clients
            symbol = 'X'
            country2digit = splits[2]
            if splits[1]=='bueno':
                yield country2digit,(symbol, 1)

    def combiner(self,key, values):
        bueno=0
        for value in values:
            if value[0] == 'A':
                yield key, ('A', value[1])
            else:
                bueno=bueno + 1

        if bueno > 0:
            yield key, ('X', bueno)

    def reducerSimple(self, key, values):
        for value in values:
            yield key,value


    def steps(self): 
        return [ 
            MRStep(mapper=self.mapper 
                   ,combiner=self.combiner
                   ,reducer=self.reducerSimple) 
        ] 


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

Я запускаю приведенный выше код так:

python mrjob-p2.py /media/notebooks/clients.csv /media/notebooks/countries.csv -r hadoop

Что дает: 

"ES"    ["X", 3]
"ES"    ["A", "Spain"]
...
"GN"    ["A", "Guinea"]
"GN"    ["X", 1]
...

Значения для ключа ES (и некоторых других) не сортируются, но для других ключей они сортируются.

Я ожидал (A должен предшествовать X, если значения были отсортированы): 

"ES"    ["A", "Spain"]
"ES"    ["X", 3]

Если я бегу локально:

python mrjob-p2.py /media/notebooks/clients.csv /media/notebooks/countries.csv -r local

Тогда я получаю: 

"ES"    ["A", "Spain"]
"ES"    ["X", 1]
"ES"    ["X", 2]
...
"GN"    ["A", "Guinea"]
"GN"    ["X", 1]
...

Что правильно.

Кто-нибудь имеет представление о том, как отсортировать значения?

Спасибо:)

...