> [!abstract] 概要(USENIX 公式ページ掲載 abstract の日本語訳) > MapReduce は、大規模データセットの処理と生成のためのプログラミングモデルおよび関連実装である。利用者は、key/value ペアを処理して中間 key/value ペアの集合を生成する map 関数と、同じ中間キーに紐づくすべての値をマージする reduce 関数を指定する。現実世界の多くのタスクがこのモデルで表現可能であることは、本論文で示すとおりである。 > > この関数型スタイルで書かれたプログラムは自動的に並列化され、コモディティマシンからなる大規模クラスタ上で実行される。ランタイムシステムは、入力データの分割、クラスタ全体にまたがるプログラム実行のスケジューリング、マシン故障の処理、必要なマシン間通信の管理といった詳細を面倒みる。これにより、並列・分散システムの経験を持たないプログラマでも、大規模分散システムのリソースを容易に活用できるようになる。 > > 我々の MapReduce 実装はコモディティマシンからなる大規模クラスタ上で動作し、高いスケーラビリティを持つ。典型的な MapReduce 計算は数千台のマシン上で数テラバイトのデータを処理する。プログラマにとってこのシステムは使いやすい。数百件の MapReduce プログラムが実装されており、Google のクラスタでは1日あたり1000件を超える MapReduce ジョブが実行されている。 ## 論文情報 - タイトル: MapReduce: Simplified Data Processing on Large Clusters - 著者・所属: [[Jeffrey Dean]]・[[Sanjay Ghemawat]]([[Google]], Inc.) - 媒体・発表年: 6th Symposium on Operating Systems Design and Implementation(OSDI '04)、USENIX Association 発行、2004年12月、San Francisco, CA - 原文 PDF: `.raw/papers/dean.pdf`(USENIX 公式アーカイブより取得) ## 概要 Google 社内で頻発していた「大量の生データ(クロール文書・Web リクエストログ等)から派生データ(転置インデックス・グラフ構造・集計等)を計算する」処理を、Lisp 等の関数型言語にある map/reduce プリミティブに着想を得た抽象化によって単純化する。利用者は map 関数と reduce 関数を書くだけでよく、並列化・耐障害性・データ分散・負荷分散の面倒な詳細はライブラリが隠蔽する。コモディティ PC 数千台のクラスタ上で高いスケーラビリティを実現し、Google の本番検索インデックス生成システムの書き換えにも利用された。 ## 問題設定 - **入力**: 大量の key/value ペア(例: クロールされた Web 文書、Web リクエストログ) - **出力**: 派生データを表す key/value ペアの集合(例: 転置インデックス、Web グラフの各種表現、ホストごとのクロールページ数集計、特定日の頻出クエリ集合) - **前提環境**: デュアルプロセッサ x86・Linux・2-4GB メモリのコモディティマシンを、100Mbps〜1Gbps のスイッチドイーサネットで接続したクラスタ。数百〜数千台規模でありマシン故障は常態。ストレージは各マシン直結の安価な IDE ディスクで、社内開発の分散ファイルシステム([[Google File System]])がレプリケーションによって可用性・信頼性を提供する。ジョブはスケジューリングシステムに投入され、タスク単位でクラスタ内の利用可能マシンに割り当てられる。 - **必要なデータ**: 入力はデフォルトで 16MB〜64MB のスプリット(ユーザーが調整可能)に分割される。 ## 提案手法 ### プログラミングモデル 利用者は次の 2 つの関数の型シグネチャに従って実装する。 ``` map (k1, v1) → list(k2, v2) reduce (k2, list(v2)) → list(v2) ``` - **Map**: 入力ペアを受け取り、中間 key/value ペアの集合を生成する。MapReduce ライブラリは同じ中間キー `I` に紐づく全ての中間値をグループ化し、Reduce 関数に渡す。 - **Reduce**: 中間キー `I` と、それに対応する値の集合を受け取り、マージしてより小さな値集合(典型的には 0 個か 1 個)を生成する。中間値はイテレータ経由で供給されるため、メモリに収まらない大きさのリストも扱える。 - 入力キー/値は出力キー/値と異なるドメインから取られるが、中間キー/値は出力キー/値と同じドメインを共有する。C++ 実装では、文字列がユーザー定義関数との間で受け渡され、文字列と型の変換はユーザーコード側の責務となる。 ### アーキテクチャ Google の計算環境向けの実装(Figure 1「Execution overview」)は次のフローで動作する(Figure 1 では番号 (1)〜(6) がフローに対応)。 **Figure 1: MapReduce の実行フロー全体像** ![[_attachments/dean/fig01-execution-overview.png]] (Figure 1. User Program が M 個の map タスク・R 個の reduce タスクを worker に fork する。(1) User Program が Master と worker 群を fork、(2) Master が idle な worker に map/reduce タスクを割り当て、(3) map worker が入力 split を読み込み、(4) 中間 key/value ペアをパーティション分割してローカルディスクに書き込み、(5) reduce worker が RPC でリモートの中間データを読み込み、(6) reduce worker が最終出力ファイルに書き込む。Source: Figure 1.) 1. MapReduce ライブラリが入力ファイルを M 個の分割(典型的に 16MB〜64MB/片)に分け、クラスタ上でプログラムのコピーを多数起動する。 2. コピーの1つが特殊な役割の **master**(マスタ)であり、残りは master から仕事を割り当てられる **worker**(ワーカー)である。map タスクが M 個、reduce タスクが R 個あり、master は idle な worker にそれぞれのタスクを割り当てる。 3. map タスクを割り当てられた worker は対応する入力 split を読み込み、key/value ペアを解析してユーザー定義の Map 関数に渡す。生成された中間 key/value ペアはメモリ上にバッファされる。 4. バッファされたペアは定期的にローカルディスクへ書き出され、パーティション関数により R 個の領域に分割される。この位置情報は master に報告され、reduce worker へ転送される。 5. reduce worker は master からの通知を受けて、map worker のローカルディスクから RPC でデータを読み込む。全中間データを読み込んだ後、中間キーでソートする(同一キーの出現をまとめるため)。メモリに収まらない場合は外部ソートを用いる。 6. reduce worker はソート済みの中間データを走査し、ユニークな中間キーごとにキーと対応する値集合をユーザーの Reduce 関数に渡す。Reduce 関数の出力は、このパーティション用の最終出力ファイルに追記される。 7. 全ての map タスクと reduce タスクが完了すると、master はユーザープログラムを起こし、MapReduce 呼び出しがユーザーコードに復帰する。 出力は R 個のファイル(reduce タスクごとに1個)として得られ、通常は1ファイルに結合する必要はない(次の MapReduce 呼び出しへの入力として利用できる)。 ### 耐障害性 - **Worker 障害**: master は各 worker を定期的に ping する。一定時間応答がなければ故障とみなす。故障した worker が完了した map タスクは idle 状態に戻され再スケジュール対象になる(出力がローカルディスク上にあり、故障で読めなくなるため)。完了済みの reduce タスクは出力がグローバルファイルシステムにあるため再実行不要。実例として、80台規模のワーカー群がネットワークメンテナンスで数分間到達不能になった際も、master が再実行を続けて計算を完遂した(§3.3)。 - **Master 障害**: master のデータ構造は定期的にチェックポイントされ、理論上は再起動可能。ただし master は単一であり故障確率が低いため、現行実装は master 故障時に計算全体を中断し、クライアントが再試行できるようにするに留めている。 - **障害時のセマンティクス**: map/reduce 関数が入力に対して決定的である場合、分散実装の出力は非障害逐次実行と同一になることを保証する。これは map/reduce タスク出力のアトミックコミット(一時ファイルへの書き込み + アトミックリネーム)によって実現される。非決定的な演算子の場合はより弱いが合理的なセマンティクスを提供する(reduce タスク `R1` と `R2` がそれぞれ異なる map 実行の出力を読む可能性がある)。 ### 局所性最適化 ネットワーク帯域は希少資源であるため、GFS が入力データをクラスタ内マシンのローカルディスクに(64MB ブロック単位、典型3レプリカで)格納している事実を活用する。master は入力ファイルの位置情報を考慮し、対応する入力データのレプリカを持つマシン(次善策として同一ネットワークスイッチ配下のマシン)に map タスクをスケジュールする。大規模な MapReduce 演算をクラスタの主要部分で実行する場合、ほとんどの入力データはローカルに読み込まれネットワーク帯域を消費しない。 ### タスク粒度とバックアップタスク(straggler 対策) - M・R は worker マシン数より十分大きくすることで、動的負荷分散と障害時の回復高速化(完了済み多数の map タスクを他の全 worker に分散できる)を実現する。master は O(M+R) のスケジューリング決定と O(M×R) の状態(map/reduce タスクペアあたり約1バイト)を保持するため実用上の上限があり、典型的には M=200,000、R=5,000、worker 2,000台という構成で運用される。 - **straggler**(処理完了が異常に遅いワーカー)が総実行時間を伸ばす主要因である。原因例: ディスク不良による読み取り速度低下(30MB/s→1MB/s)、他タスクとのリソース競合、プロセッサキャッシュ無効化バグ(処理速度が100倍超悪化した実例)。 - 対策として、MapReduce 演算が完了に近づくと master が残りの実行中タスクの**バックアップ実行**をスケジュールする。一次実行・バックアップ実行のいずれか早く完了した方をもって完了とする。この機構による追加計算資源消費は数%程度に抑えられている。 ### その他の拡張(Refinements) - **パーティション関数**: 既定は `hash(key) mod R` によるほぼ均等な分割。ユーザーが独自関数を指定でき、例えば `hash(Hostname(urlkey)) mod R` で同一ホストの URL を同一出力ファイルにまとめられる。 - **順序保証**: 与えられたパーティション内で中間 key/value ペアは増加キー順に処理される。 - **Combiner 関数**: reduce 関数が可換・結合的な場合、map タスク側で部分マージを行うオプション。単語数カウント例では Zipf 分布により同一語(例: `<the, 1>`)が大量発生するため、ネットワーク送信前に部分和を取ることで帯域を節約する。 - **入出力型**: text モード(行オフセットをキー、行内容を値とする)などの標準リーダー/ライター実装に加え、DB やインメモリ構造からの読み込みにも対応するリーダーインターフェースをユーザーが実装可能。 - **副作用ファイル**: map/reduce 演算子が補助的な出力ファイルを生成する場合、アプリケーション側でアトミック性・冪等性を担保する必要がある(2フェーズコミットは未サポート)。 - **不正レコードのスキップ**: セグメンテーション違反等でクラッシュするレコードを検知し、決定的クラッシュと判定されたレコードは以後のスキップ対象にできる(シグナルハンドラ + UDP による master への通知)。 - **ステータス情報**: master 内蔵の HTTP サーバがタスク進捗・処理率・失敗した worker などを可視化するステータスページを提供する。 - **カウンタ機構**: ユーザーコードが named counter を作成し、map/reduce 内でインクリメントすることでイベント発生数を追跡できる。master が重複実行分を除いて集計する。 ## 新規性 先行研究(並列プレフィックス計算、Bulk Synchronous Programming、MPI の高レベル抽象等)は計算の並列化を自動化する制約付きプログラミングモデルを提供してきたが、多くは小規模実装にとどまり、マシン故障の処理の詳細はプログラマに委ねられていた。MapReduce はこれらのモデルの単純化・蒸留と位置づけられるが、数千プロセッサ規模までスケールする**耐障害性を備えた実装**を提供する点で異なる。局所性最適化は active disks の思想([12][15])に着想を得るが、専用のディスクコントローラではなく少数のディスクを直結したコモディティプロセッサ上で実現している。バックアップタスク機構は Charlotte System の eager scheduling([3])に類似するが、単純な eager scheduling の弱点(あるタスクが繰り返し失敗すると計算全体が完了しない)を不正レコードスキップ機構で緩和する。River([2])は異種ハードウェアや性能変動下での平均ケース性能をディスク・ネットワーク転送の慎重なスケジューリングで実現するのに対し、MapReduce は制約付きプログラミングモデルにより問題を大量の細粒度タスクへ分割し、動的スケジューリングと冗長実行(バックアップタスク)によって非均一性の影響を緩和するという異なるアプローチを取る。 ## 実験設定 - **環境**: 約1800台のマシンからなるクラスタ。各マシンは 2GHz デュアル Intel Xeon(Hyper-Threading 有効)、4GB メモリ、160GB IDE ディスク2台、ギガビットイーサネットリンクを備え、二層ツリー型スイッチネットワーク(ルート集約帯域 約100-200Gbps)で接続。同一ホスティング施設内で全マシン間の往復遅延は1ミリ秒未満。4GB メモリのうち約1-1.5GB は他タスクに予約済み。実験は CPU・ディスク・ネットワークが概ねアイドルな週末午後に実行。 - **評価タスク**: (1) Grep — 約1テラバイトのデータ(10^10件・100バイトレコード)から出現頻度92,337件のまれな3文字パターンを検索。入力は約64MBピース(M=15,000)に分割、出力は1ファイル(R=1)。(2) Sort — 約1テラバイト(10^10件・100バイトレコード)をソート。TeraSort ベンチマークをモデルにした約50行未満のユーザーコード。M=15,000、出力を2-way レプリケーションで4,000ファイル(R=4,000)に分割。(注: PDF テキスト抽出でスーパースクリプトが失われ「1010」と表示されるが、100バイト×レコード数=約1テラバイトという本文記述と整合するのは10^10件であるため、本ページでは10^10と表記する) - **比較対象**: Sort では現行最良報告値である TeraSort ベンチマークの1057秒([18])と比較。 ## 実験結果 **Figure 2: grep 実行時のデータ転送レート** ![[_attachments/dean/fig02-grep-data-transfer-rate.png]] (Figure 2. 入力読み取りレートは 1764 worker 割り当て時に 30GB/s 超でピークに達し、map タスク完了とともに約80秒でゼロに落ちる。計算全体は起動オーバーヘッド(GFS への1000入力ファイルオープン等、約1分)を含め約150秒で完了。Source: Figure 2.) - **Sort(通常実行)**: 入力読み取りレートは約13GB/sでピーク、全 map タスクが200秒未満で完了。shuffle は最初の map タスク完了と同時に開始し、約600秒で全シャッフルが完了。出力書き込みは約2-4GB/sで継続し、約850秒で全書き込み完了。起動オーバーヘッドを含め全体で891秒。TeraSort の現行最良報告値1057秒に近い水準([18])。入力レートが shuffle レート・出力レートより高いのは局所性最適化(ローカルディスク読み取りがネットワークをバイパス)のため。出力はレプリケーション(2コピー)のため shuffle レートより低い。 **Figure 3: sort プログラムの実行別データ転送レート推移(a: 通常実行 b: バックアップタスク無効 c: 200タスク強制終了)** ![[_attachments/dean/fig03-sort-data-transfer-rates.png]] (Figure 3. (a) 通常実行では891秒で完了。(b) バックアップタスクを無効化すると、960秒時点で reduce タスクの5個を除き完了するが、残り5個のストラグラーの完了に300秒を要し、全体で1283秒(44%増)を要した。(c) 実行途中で1746台中200 worker プロセスを強制終了しても、クラスタスケジューラが即座にプロセスを再起動し、933秒(通常比5%増)で完了した。Source: Figure 3.) - **バックアップタスクの効果**: バックアップタスク無効化により sort の完了時間が44%増加(891秒→1283秒)。長いテールは僅かなストラグラー(reduce タスク5個)によって生じる。 - **ワーカー障害耐性**: 1746 worker プロセス中200個を意図的に強制終了しても、入力レートは一時的に負(map 出力の再計算)になるが、933秒(通常実行比5%増)で完了。MapReduce は大規模ワーカー障害に対して頑健である。 **Figure 4: MapReduce インスタンス数の推移** ![[_attachments/dean/fig04-instances-over-time.png]] (Figure 4. Google のソースツリーにチェックインされた個別 MapReduce プログラム数は、2003年初頭の0件から2004年9月末までに約900件近くまで成長。Source: Figure 4.) **Table 1: 2004年8月に実行された MapReduce ジョブの統計** ![[_attachments/dean/table01-jobs-august-2004.png]] (Table 1. 2004年8月のジョブ数は29,423件、平均ジョブ完了時間634秒、使用マシン日数79,186日、入力データ読み取り量3,288TB、中間データ758TB、出力データ193TB、ジョブあたり平均worker数157台、平均worker故障数1.2件、平均mapタスク数3,351件、平均reduceタスク数55件、ユニークなmap実装395種・reduce実装269種・map/reduce組み合わせ426種。Source: Table 1.) - **本番検索インデックス生成システムへの適用(§6.1)**: 20テラバイト超のクロール文書を入力とし、5〜10段のMapReduce演算からなる処理列としてインデックス生成システムを全面書き換え。旧来の場当たり的な分散パスと比較して、ある計算フェーズのコードサイズが約3,800行の C++ から約700行に削減された。 ## 考察 MapReduce の成功は3つの要因に帰着される。第一に、並列・分散システムの経験がないプログラマでも使いやすいこと(並列化・耐障害性・局所性最適化・負荷分散の詳細を隠蔽するため)。第二に、多様な問題(検索インデックス生成、ソート、データマイニング、機械学習、大規模グラフ計算等)がMapReduce計算として容易に表現できること。第三に、数千マシン規模にスケールする実装を開発できたこと。制約付きプログラミングモデルの採用が、並列化・耐障害計算を容易にする鍵であり、ネットワーク帯域の希少性が局所性最適化と中間データのローカルディスク保存という設計判断を導いた。冗長実行(バックアップタスク)は遅いマシンの影響緩和とマシン故障・データ損失への対処の両方に有効であることが示された。 ## 強み / 弱点・課題 - **強み**: シンプルな2関数インターフェースでの高い表現力(distributed grep・カウント・逆リンクグラフ・term vector・inverted index・distributed sort 等、多様なタスクを容易に記述可能)。数千マシン規模での実運用実績(2004年8月時点で月29,423ジョブ、入力3,288TB)。決定的関数に対する逐次実行との出力等価性保証。バックアップタスク機構による straggler 耐性。デバッグ用のローカル逐次実行モードやステータスページによる運用性の高さ。 - **弱点・課題**: master が単一障害点であり、master 故障時は計算全体を中断する設計(チェックポイントからの再起動機構は実装されているが、実運用では単純に中断・再試行に頼る)。2フェーズコミットをサポートしないため、複数出力ファイル間の整合性が必要なタスクは決定的である必要がある。パーティション数・タスク数の実用上限(master の O(M+R) スケジューリング決定と O(M×R) メモリ状態)。ネットワーク帯域節約のためレプリケーションを用いており、消去符号化(erasure coding)を採用すればさらに帯域を削減できる可能性がある、と論文自身が指摘している。