現在JavaScriptの設定が無効になっています。このページを利用するにはJavaScriptの設定を有効にしてください。引き続き、MapReduceのカスタマイズポイントを解説していきます。前回の記事の図や、表などを参照しながら読み進めていただければと思います。MapperやReducerの流れの制御Mapperの実行の流れは、デフォルトでは、初期化処理を行った後、map関数を繰り返し実行し、終了処理を行うようになっていますが、この流れ自体を制御することができます。古いAPIでは、MapRunnerを通じてこの流れを制御できますが、0.20.0からの新しいAPIでは単純にMapperクラスのrun関数をオーバーライドすることで、行えます。デフォルトのrun関数は以下の通りです。}初期化処理を行い、map関数を繰り返し実行し、終了処理する流れがそのまま書かれています。WEBページのクロールなど、一回のmap関数の実行に時間がかかる処理の場合、map関数をさらに複数スレッドで実行した方が効率的です。した場合に、このrun関数を活用できます。Reducerの流れの制御も同様にrun関数を上書きすることで行えますが、新しいAPIでしか行えず、古いAPIではMapRunnerに相当するものがありません。PartitionerPartitionerはmap関数の出力を、Partitionに分割し、それぞれをReduceTaskに割り当てます。デフォルトのPartitionerはHashPartitionerで、Keyのハッシュ値にもとづいて分割します。}}getPartition関数は、Keyのハッシュ値をReduceTaskの数で割った余りを返し、Partitionを決定します。同じKeyは必ず同じReduceTaskで処理されることになります。Partitionerをカスタマイズする例をあげてみます。map関数の出力Keyが、ファイル名だったとして、同じ拡張子のファイルは、同じReduceTaskで処理したいとしましょう。同じ拡張子のファイルは必ず同じReduceTaskで処理されるというわけです。その後のReducerでは、同じ拡張子のファイルの長さの合計を計算する、といった処理が考えられますが、それには、もう一工夫必要です。reduce関数に渡される、Valueのグルーピングを拡張子ごとに行う必要がありますが、これは後ほど解説します。CombinerCombinerは、Mapの出力をさらに集計する処理のことでMapTask内で行われます。Combinerの新たな出力が、Reducerの入力として渡されます。解説したWordCountの例をあげてみましょう。Combinerを使わないと以下のようになります。map関数の出力のKeyに単語の名前、Valueには1がセットされ、map関数の出力がそのままreduce関数に渡されます。単語の数の合計を出力します。combinerを使うとどうなるでしょうか?map関数の出力は、MapTask内でCombinerに渡されます。Combinerでは、そのMapTask内での単語の数の合計を出力し、reduce関数に渡します。reduce関数では、さらにそれらを足し合わせて、全体の単語の数の合計を出力するというわけです。Combinerは、ReduceTaskへの転送量を減らし、Reducerの計算量を減らすことで、全体の高速化を図る目的で使われます。Combinerはオプションであり、必ずしも使うべきではないケースも多いので注意して下さい。DistributedCacheDistributedCacheは、処理に必要なファイルを各マシンにコピーしてキャッシュする仕組みです。配布は、Taskが実行される前に、一回だけ行われます。キャッシュしたファイルは、Mapper、Reducerから自由に利用することができますが、読み込み専用で、変更してはいけません。ファイルをキャッシュするだけではなく、zip/tar/tgz/tar.gzなどの圧縮されたファイルは展開してキャッシュされます。jarファイルを配布すると、自動でクラスパスに追加してくれ、MapperやReducerで使うことができます。.soなどのネイティブライブラリも同様に配布できます。jarファイルを配布する例は以下の通りです。jobはJobConfファイルなお、MapperやReducerで共通で扱いたい文字列などは、JobConfに直接定義してしまうのが便利です。);MapやReduceの初期化処理などで、以下のようにしてJobConfに定義した文字列を取得できます。}}ComparatorComparatorは、Keyのソートや、ReducerでのValueのグルーピングをカスタマイズできます。Key同士をどう比較するかを定義します。Keyのソートや、Valueのグルーピングに用いる、RawComparatorのインターフェイスは以下です。l2);}Keyのソートや、Valueのグルーピングでは、このcompare関数が呼ばれてバイト単位で比較されることになります。RawComparatorのデフォルトの実装は、WritableComparatorです。a.compareTo(b);}WritableComparatorでは、バイト列からKeyを抽出し、そのKeyのクラスのcompareTo関数を用いて比較するようになっています。カスタマイズしたComparatorを、Valueのグルーピングに用いる例をあげてみます。先ほどのPartitionerのときの例では、同じ拡張子のファイルを同じReduceTaskに割り当てる処理を行いました。その後のReducerでは拡張子ごとにファイルのバイト数の合計などを求める、などの処理が考えられますが、その場合、拡張子ごとにValueをグルーピングする必要があります。拡張子ごとにグルーピングするComparatorクラスは以下の通りです。Keyにはファイル名がセットされています。}}まず比較したいKeyのクラス(Text)を指定して、WritableComparatorのコンストラクタを呼び出します。比較する際にcompare関数が呼び出されるので、ファイル名から拡張子を抽出して、TextのcompareToメソッドを呼び出しています。このクラスをJobConfのsetOutputValueGroupingComparator関数で指定すれば、拡張子ごとにグルーピングされるというわけです。実はこの方法だと、一度Keyのバイト列をデシリアイズしてから比較しているので、低速です。MapReduceでは、この比較のスピードは全体の処理時間に大きく影響します。TextのデフォルトのComparatorでは、バイト列をそのまま比較する最適化を行っています。}}WritableComparatorのcompareBytesクラスを呼び出し、バイト列のまま比較を行っています。Textは、文字列のバイトの長さを、先頭に可変長で格納しています。そこで、WritableUtilsクラスのdecodeVIntSize関数で、長さの情報を格納しているヘッダ自体のバイト数を求めて、そのヘッダを除外して比較を行っています。Comparator());}のようにstaticブロックで、WritableComparatorのdefine関数を呼び出します。OutputFormatOutputFormatのインターフェイスは以下のようになっています。IOException;}InputFormatと同様に、getRecordWriter関数で、KeyとValueを受け取って実際にデータを出力するRecordWriterクラスを返します。RecordWriterのインターフェイスは以下です。IOException;}write関数が繰り返し呼ばれて、データが書き込まれます。Key/ValueをXMLに変換して書き出したい、HDFSではなくDBに書き込みたい、などの場合にOutputFormatのカスタマイズは有効です。いう便利な仕組みも用意されています。WritableWritableクラスはMapReduceのKeyやValueとして使われるクラスです。テキストにはTextクラス、intにはIntWritableといった具合にWritableクラスが用意されています。Writableクラスは、マシン間で転送可能なバイトストリームへのシリアライズ、デシリアライズを行います。Javaで用意されているものより、コンパクトで高速に処理できるのが特徴です。Writableクラスのインターフェイスは以下の通りです。out);}readFields関数はバイトストリームのDataInputを読み取り、デシリアイズを行います。write関数は、シリアイズを行い、バイトストリームのDataOutputに書き出します。緯度経度を扱うWritableクラスは以下の通りです。MapReduceのKeyとして扱うためには、比較も可能であるようにWritableComparableインターフェイスを実装します。}}write関数では、DataOutputのwriteDouble関数を呼び出しで、バイトストリームへの書き出しを行っています。readFields関数では、DataInputのreadDouble関数でバイトストリームから読みだし、緯度経度を設定しています。比較は、東京駅からの距離の逆順で行うようになっています。(東京駅からの距離が近い方が大きい値)コンストラクタで、東京駅の緯度経度を指定してLatLonDistanceクラスを初期化し、compareTo関数内で、東京駅からの距離を計算しています。高速化のためには前述したComparatorを実装する必要があります。Partitionerで呼び出される、hashCode関数なども必要です。次回は以上でカスタマイズポイントの解説は終わりです。次回は、実際のサービスでの活用事例をあげ、今までに解説したカスタマイズポイントがどういじられているかを紹介する予定です。JAPANでは情報技術を駆使して人々や社会の課題を一緒に解決していける方を募集しています。詳しくは採用情報をご覧ください。JAPANのサービス開発を支える技術やデザイン、文化を発信しています。Reserved.このページの先頭へ。