Карта/сокращение двухэтапного порядка подсчетов

Question

Эта программа python3 пытается создать частотный список слов из текстового файла, используя map/reduce. Я хотел бы знать, как упорядочить подсчет слов, представленный как «счет» во втором выражении выхода редуктора, так что наибольшие значения подсчета появляются последними. В настоящее время хвост результатов выглядеть следующим образом:

"0002" "wouldn" 
"0002" "wrap" 
"0002" "x" 
"0002" "xxx" 
"0002" "young" 
"0002" "zone" 

Для контекста, я прохожу любой текстовый файл слово в программу Python3 так:

python MapReduceWordFreqCounter.py book.txt 

Вот код для MapReduceWordFreqCounter.py:

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

# ignore whitespace characters 
WORD_REGEXP = re.compile(r"[\w']+") 

class MapReduceWordFreqCounter(MRJob): 

    def steps(self): 
     return [ 
      MRStep(mapper=self.mapper_get_words, 
        reducer=self.reducer_count_words), 
      MRStep(mapper=self.mapper_make_counts_key, 
        reducer = self.reducer_output_words) 
     ] 

    def mapper_get_words(self, _, line): 
     words = WORD_REGEXP.findall(line) 
     for word in words: 
      yield word.lower(), 1 

    def reducer_count_words(self, word, values): 
     yield word, sum(values) 

    def mapper_make_counts_key(self, word, count): 
     yield str(count).rjust(4,'0'), word 

    def reducer_output_words(self, count, words): 
     for word in words: 
      yield count, word 

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

Answer 1

Вы должны установить специализированный компаратор сортировки для своей работы.

Если вы написали это в Java, это будет выглядеть как

job.setSortComparatorClass(SortKeyComparator.class); 

, и вы должны предоставить класс, который дает обратный порядок

public class SortKeyComparator extends Text.Comparator { 

    @Override 
    public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { 
     return (-1) * super.compare(b1, s1, l1, b2, s2, l2); 
    } 
} 

Я предполагаю питон Hadoop апи имеет некоторые Двойники методы для этого трюка.

Answer 2

Для шага уменьшения MRJob нет ожидаемого, что результаты должны быть заказаны с помощью клавиши «count».

Здесь импорт MRJob позволяет запускать код локально и в кластере AWS Elastic MapReduce. MRJob делает тяжелую работу для выполнения, так как использует API Yarn API и Hadoop для распределенной передачи данных между картой и сокращения рабочих мест.

Например, чтобы запустить локально, вы можете запустить этот MRJob как: питон MapReduceWordFreqCounter.py books.txt> counts.txt

Для запуска на одном узле ОГО: питон MapReduceWordFreqCounter.py -r ОГО books.txt> counts.txt

Для запуска на 25 ЭХ узлах: Python MapReduceWordFreqCounter.py -r ОГО --num-EC2-экземпляры = 25 books.txt> counts.txt

Для устранения распределенного Работа EMR (замените свой идентификатор задания): python -m mrjob.tools.emr.fetch_logs --find-failure j-1NXEMBAEQFDFT

Здесь, при запуске на четырех узлах, уменьшенные результаты упорядочены, но находятся в четырех разных разделах выходного файла. Оказывается, что принуждение редуктора к созданию отдельного упорядоченного файла не имеет преимуществ по производительности, чем просто упорядочение результатов на этапе задания после запуска. Таким образом, один из способов решения этого конкретного вопроса заключается в использовании команды вроде Linux:

sort word_frequency_list.txt > sorted_word_frequency_list.txt 

, который производит эти «хвостатые» результаты:

«0970» «из» «1191» «а» " 1292" „“ „1420“ „ваш“ „1561“ „вы“ „1828“ „в“

в более общем плане, есть рамки поверх Hadoop, которые идеально подходят для этого вида обработки.Для этой проблемы Pig можно использовать для чтения в обработанном файле и упорядочить подсчеты.

Свинья могут запускаться через оболочку Grunt или с помощью скриптов Pig (с использованием синтаксиса Pig Latin с учетом регистра). Свиньи сценарии следуют следующему шаблону: 1) Несущая оператор чтения данных 2) Серия высказыванием «преобразования» для обработки данных 3) свалка/STORE заявление, чтобы сохранить результаты

Чтобы заказать отсчеты с помощью свинью:

reducer_count_output = LOAD 'word_frequency_list.txt' using PigStorage(' ') AS (word_count:chararray, word_name:chararray); 
counts_words_ordered = ORDER reducer_count_output BY word_count ASC; 
STORE counts_words_ordered INTO 'counts_words_ordered' USING PigStorage(':', '-schema');