カウントの2段階順序付けのマップ/縮小

Question

このpython3プログラムは、map/reduceを使用してテキストファイルから単語の頻度リストを生成しようとします。私は、最大のカウント値が最後に現れるように、第2レデューサーの利回り明細書で「カウント」として表される、ワード数をどのように並べるかを知りたいと思います。現在、結果の尾は次のようになります。コンテキストの場合

"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); 

ようになり、あなたは私がPythonのHadoopのAPIは、いくつかのまねを持っていると思い逆順に

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); 
    } 
} 

を与えるクラスを提供する必要がありますこのトリックを行う方法。

Answer 2

MRJob Reduceステップでは、結果をキー 'count'によって順序付ける必要はありません。

ここでは、MRJobのインポートにより、コードをローカルで、AWS Elastic MapReduceクラスタで実行することができます。 MRJobは、マップAPIと縮小ジョブ間の分散データ転送用にYarn APIとHadoopストリーミングを使用するため、実行に重いものがあります。単一EMRノード上で実行するには

counts.txt パイソンMapReduceWordFreqCounter.py books.txt>：たとえば、ローカルで実行するために、あなたは、このMRJobを実行することができます

パイソンMapReduceWordFreqCounter.py -r EMR books.txt>

counts.txt 25個のEMRノード上で実行する： パイソンMapReduceWordFreqCounter.py -r EMR --num-EC2-インスタンス= 25 books.txt>

counts.txt分散をトラブルシューティングしますEMRジョブ（職種を代用）： python -m mrjob.tools.emr.fetch_logs --find-failure j-1NXEMBAEQFDFT

ここで、4つのノードで実行している場合、縮小された結果は出力されますが、出力ファイルでは4つのセクションに分かれています。レデューサーを単一の順序付きファイルを生成するように強制することは、実行後のジョブステップで結果を順序付けるだけではパフォーマンス上の利点がありません。

sort word_frequency_list.txt > sorted_word_frequency_list.txt 

これらの「尾行」の結果生成します：

「0970」「」 「1191」の「」 」をこのように、この特定の問題を解決する一つの方法は、Linuxのコマンドのソートを使用することです1292" 『』 『1420』 『あなたの1561年『』 』 『君』 『1828』 『に』

より一般的には、処理のこの種のための理想的なのHadoopの上にフレームワークがあります。この問題のために、Pigは処理されたファイルの読み込みとカウントの順序付けに使用できます。

PigはGruntシェルまたはPigスクリプト（大文字と小文字を区別するPig Latin構文を使用）で実行できます。豚のスクリプトは、次のテンプレートに従ってください： 1）LOAD文は豚を使用してカウントを注文する結果

を保存するためのデータ 3）DUMP/STORE文を処理するために）「変革」ステートメントの一連のデータ 2を読み取るために：

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');