MapReduce論文を読む

はじめに

MapReduceは、Googleが以前提案した、大規模なデータセットの並列計算を支えるソフトウェアアーキテクチャのモデルです。現在では、多くの分散システムでこの考え方が使われています。

その理論は、Googleが2004年に発表した論文『MapReduce: Simplified Data Processing on Large Clusters』にまとめられています。全文はこちらで読めます。わずか13ページですが、その辺の短い論文とは比べものにならないほど情報が詰まっています。

論文を読みながら書いたメモなので、少しまとまりに欠けるかもしれません。

プログラミングモデル

MapReduceは、とてもシンプルな並列処理モデルです。フレームワークを利用する側が指定するのは、次の2つの関数だけです。

Map関数：1つのキーと値のペアから、一連の中間的なキーと値のペアを生成する。
Reduce関数：同じキーを持つ中間値をまとめる。

データの分配、タスクの割り当て、エラー処理、負荷分散など、残りの細かい処理はフレームワークが引き受けます。利用者はそれらを詳しく知らなくても、アプリケーションのロジックに集中できます。

大まかな処理の流れは次のとおりです。

Mapは入力のキーと値のペアを受け取り、一連の中間的なキーと値のペアを生成します。MapReduceフレームワークは、中間キーが同じ値を集めてReduce関数に渡します。Reduce関数は中間キーとその値の集合を受け取り、通常はそれらをより小さな集合に集約します。1回の呼び出しで結果を1つだけ返す場合もあれば、何も返さない場合もあります。

大量のテキストに含まれる単語を数える例です。

map(String key, String value):
    // key：文章名称
    // value：文章内容
    for 单词 w in value:
        增加中间计数 (w, "1")
 
reduce(String key, Iterator values):
    // key：一个单词
    // value：一系列计数
    int result = 0;
    for v in values:
        result += ParseInt(v);
    输出 (ToString(result))

実装

実行の流れ

MapReduceはプログラミングモデル、あるいはプログラミングの考え方なので、実装方法はさまざまです。Googleの論文では、ローカルネットワークでつながった多数のマシン向けの実装が紹介されています。実行の流れは次の図のとおりです。

MapReduceの実行フロー

まず、MapReduceフレームワークが入力ファイルをM個に分割します。各分割の大きさは通常16〜64MBです。その後、クラスタ内のマシン、つまりプロセスを起動します。
クラスタ内の1つのプロセスが、特別なmasterプロセスになります。残りのworkerプロセスには、masterがタスクを割り当てます。割り当てるタスクはM個のmapタスクとR個のreduceタスクです。masterは空いているworkerを選び、mapまたはreduceタスクを1つずつ渡します。
mapタスクを割り当てられたworkerは、対応する入力の分割を読み、キーと値のペアを取り出して、ユーザー定義のmap関数に順に渡します。map関数が返した中間的なキーと値のペアは、いったんメモリに蓄えます。
メモリに蓄えたペアは、分割関数でR個の領域に分け、定期的にローカルディスクへ書き出します。ディスク上の保存場所はmasterに通知し、masterがreduceタスクを担当するworkerに転送します。
reduce workerはmasterから保存場所を受け取ると、データを保持しているmap workerにRPCリクエストを送り、データを読み出します。すべての中間データを読み終えたら、キーでソートして同じキーの値をまとめます。1つのreduceタスクで異なるキーを多数扱うため、このソートが必要です。データが大きすぎる場合は、外部ソートを使うこともあります。
reduce workerはソート済みの中間データを走査し、それぞれのキーと対応する値の集合をユーザー定義のreduce関数に渡します。reduce関数の出力は、最終的な出力ファイルに追記します。出力ファイルはreduceの分割ごとに1つです。
すべてのmapタスクとreduceタスクが終われば、MapReduceの処理は完了です。

結果はR個の出力ファイルに保存されます。これらのファイルを、次のMapReduce処理の入力にすることもよくあります。

耐障害性

ここではworkerが落ちた場合だけを考えます。masterの障害まで扱うと、選挙や合意形成など複雑な話になるためです。

masterとworkerの間ではハートビートをやり取りし、一定時間workerから応答がなければ、そのworkerは落ちたと判断します。そのworkerが完了したmapタスクはすべて未着手に戻し、別のworkerへ割り当て直します。障害発生時に実行中だったmapまたはreduceタスクも、未着手に戻します。

完了済みのmapタスクを再実行するのは、出力が障害の起きたマシンのローカルディスクにあるからです。一方、完了済みのreduceタスクの出力はグローバルなファイルシステムに置かれるため、再実行は不要です。

あるmapタスクを最初はAが実行していたものの、Aが落ちてBに再割り当てされたとします。この情報はreduceタスクを実行するすべてのworkerに通知されます。まだAからデータを読み取っていないreduceタスクは、代わりにBから読み取るようになります。

マシンの性能が低くても、ネットワークが正常なら障害とは判定されません。すると、その遅いマシンがシステム全体の足を引っ張り、ほかの処理が終わってもその完了を待たされます。Googleの実装では、MapReduce処理全体が終わりに近づいたところで、まだ実行中のタスクを空いている別のworkerにも割り当てます。元のworkerと追加で割り当てられたworkerのどちらかが終われば、そのタスクは完了とみなします。

性能改善や細かな最適化、拡張については省略します。