MapReduce:Simplified Data Processing on Large Clusters

・モチべーション
　巨大なデータを処理したい
　　多くのタスク：ある多くのデータから，新しいデータを生成する
　　数百の CPU を使う
　　しかも，簡単に
　MapReduce の機能
　　自動的な並列分散処理
　　耐障害性
　　I/O スケジューリング
　　状態のモニタリング

・プログラミングモデル
　入出力は Key/Value のペア
　2つの関数を定義
　　map (in_key, in_value) -> list(out_key, intermediate_value)
　　　key/value ペアを処理
　　　媒介としてのペアのセットを生成
　　reduce (out_key, list(intermediate_value)) -> list(out_value)
　　　全ての媒介ペアのセットの結合
　　　マージされたある一つのセットを生成(大体一つ)
　LISP中の似たようなプリミティブや他の言語にインスパイアされた

・サンプル：単語頻度のカウント

　map(String input_key, String input_value):
　　// input_key　: 文書名
　　// input_value: 文書
　　for each word w in input_value:
　　　EmitIntermediate(w, "1");

　reduce(String output_key, Iterator intermediate_values):
　　// output_key　 :単語
　　// output_values:頻度のリスト
　　int result = 0;
　　for each v in intermediate_values:
　　　result += ParseInt(v);
　　Emit(AsString(result));

・広く使われるモデル
　2003年5月あたりから使われ始めて(？)，1年半で900ソース程度に使われている
　例えば
　　分散grep/sort
　　Webリンクの逆解析(web link-graph reversal)
　　ホスト単位の語ベクトル
　　Webアクセスログの統計
　　逆索引の生成
　　文書クラスタリング
　　機械学習
　　統計的機械翻訳

・実装の概要
　通常のクラスタ
　　数百/数千の 2-CPU x86マシン，2-4GBのメモリ
　　限定的な二分バンド幅
　　ストレージはローカルなIDEディスク
　　GFS(Google File System)
　　ジョブスケジューリングシステム：スケジューラによるマシンへのタスクの割り当て
　C++のライブラリとして実装

・実行/並列実行
　複数の入力(Map)について，同じkeyを持つ value をまとめて，出力(Reduce)
　複数の Map タスクによる入力について，Partitioning Function にて Key による Reduce タスクへの振り分けを行う(注：図を見る限りはそうだが，真偽は不明)

・タスクの粒度とパイプライン
　ちょうど良い粒度のタスク：マシン数よりマップタスクの方が多い
　　失敗の際のリカバリー時間の最小化
　　Map実行時のパイプラインをシャッフルを可能にする
　　より良い動的なロードバランシング
　普段，200,000 Map/5000 Reduce タスクが，2000台のマシン上で使われる

・MapReduce の管理画面
　まず最初の30分で，878934.6MB(858GB)のMap処理，523389.6MB(511GB)のReduceへの入力が終了
　次の10分で，Reduce 処理が終了

・耐障害性：再実行による解決
　実行時の障害について
　　定期的なハートビートメッセージによる，障害の検出
　　完了した，あるいは実行中の Map タスクの再実行
　　実行中の Reduce タスクの再実行
　　マスターを通して，タスクの完了がコミットされる
　マスターの障害について
　　まだ扱えない(ほとんど起きない)
　ロバスト性：1800台のうち1600台が一旦ロストしても，タスクは完了する
　障害があったときの動作 → 論文を参照

・改良点：冗長な実行
　遅いワーカーは著しく実行時間を長くする
　　マシン上でリソースを消費する他のジョブ
　　ソフトエラーを伴う，良くないディスクはデータの転送レートが極めて低い
　　プロセッサのキャッシュが使えないといった珍しいこと
　解決方法：段階の終了間際に，タスクのコピーを実行する
　　どちらか先に終われば，それで良し
　効果：急激にジョブの処理時間を短くした

・改良点：局所最適化
　マスターのスケジューリングポリシー
　　入力ファイルブロックのレプリカの所在をGFSに尋ねる
　　Map タスクは通常，(GFS のブロックサイズと同じ)64MB単位に区切られる
　　GFS の入力ブロックのレプリカが同じマシン/ラックにあることで，スケジューリングされたMap タスク
　効果：ローカルディスクを読み出すのと同じ位の速さで，数千のマシンが入力を読み出せる
　　これがなければ，スイッチは読み込みレートを制限されていた

・改良点：不具合のあるレコードのスキップ
　Map/Reduce は，ときどき特定の入力によって障害が発生する
　　もっとも良い解決法は，ちゃんと debug と fix を行うことだけど，実際には無理．
　セグメントの障害に対して
　　シグナルハンドラーからマスターに対して，UDP パケットを送信する
　　処理中のレコードのシーケンス番号を含む
　マスターが同じレコードに2つの不具合を発見したら
　　次のワーカーは，そのレコードのスキップを指示される
　効果：第3者によるライブラリ中の bug に対処できる

・他の改良点(論文を参照)
　Reduce パーティション毎に，保証順位をソートする
　媒介データの圧縮
　結合器(Combiner)：ネットワーク帯域の節約に貢献
　debug/testのための，ローカル環境での実行
　ユーザ定義の計算

・パフォーマンス
　1800台のクラスタ上でテスト実行
　　4GBのメモリ
　　2GHz Xeon dual-processor(HyperThreading)
　　Dual 160GB IDEディスク
　　100Gbps相当の二分バンド幅
　2つのベンチマーク
　　MR_Grep：まれなパターンにマッチするレコード(92Kレコード)を抽出するために，10^10 100-byteのレコードからスキャン
　　MR_Sort：10^10 100-byteのレコードをソート(TeraSortをモデルとする)

・MR_Grep
　実行後，60秒付近まではスループットが向上し，また次第に減少する
　局所最適化の効果
　　1800台のマシンは1TBのデータを読み，ピークは31GB/s近くになった
　　これがなければ，スイッチは10GB/sが限界
　起動時のオーバヘッドは，短時間のジョブでは深刻である

・MR_Sort
　処理時間
　　ノーマル　　　　　　　： 839秒
　　バックアップタスクなし：1235秒
　　200プロセッサをkill　 ： 886秒
　バックアップタスクは，特にジョブ処理時間を減らす
　　システムは，障害を上手に扱える

・Google のインデキシング・システムの書き直し
　MapReduce を使ったシステムの書き直しが行われている
　　24 MapReduce Operations のセット
　　新しいコードはよりシンプルに，より理解が簡単に
　　MapReduce は，障害や遅いマシンに善処する
　　マシンの追加によってインデクスをより速くするのが，簡単になる

・利用状況：2004年8月の MapReduceジョブ

　Number of jobs　　　　　　　　　29,423　　
　Average job completion time　　　　634 secs　　
　Machine days used　　　　　　　 79,186 days　　
　Input data read　　　　　　　　　3,288 TB　　
　Intermediate data produced　　　　 758 TB　　
　Output data written　　　　　　　　193 TB　　
　Average worker machines per job　　157　　
　Average worker deaths per job　　　1.2　　
　Average map tasks per job　　　　3,351　　
　Average reduce tasks per job　　　　55　　
　Unique map implementations　　　　 395　　
　Unique reduce implementations　　　269　　
　Unique map/reduce combinations　　 426

・関連技術
　関数型言語のプレミティブな部分にインスパイアされたプログラミングモデル
　たくさんの巨大なソートシステムに類似したパーティショニング/シャッフリング
　　NOW-Sort ['97]
　耐障害性のための再実行
　　BAD-FS['04] and　TACC['97]
　Active Disks/Diamond と一体となった並列処理による局所最適化
　　Active Disks['01], Diamond ['04]
　Charlotte system の中の Eager Scheduling に似たバックアップタスク
　　Charlotte ['96]
　River の分散キューと同様に似た問題を解決した，動的なロードバランシング
　　River ['99]

・結論
　MapReduce は，有効な概念となっていることを証明している
　Google での巨大な計算を非常に単純にする
　問題にフォーカスし，面倒な細かいことを片付けるようなライブラリにしていく

---------------

[関連サイト]
近況(2006-01-27)
[メモ]GoogleのMapReduceの勉強会の様子
MapReduce: 大規模クラスタでの簡単なデータ処理 (#318th PTT) - Ogawa::Memoranda
巨大な検索システムを耐障害性の高いソフトと安価なマシンで実現 ← 2004年に紹介されていたんですね
たまには : スタッフの戯言
ITmediaニュース：Google検索システムの舞台裏 (2/2)
いま明かされる、グーグル・データセンターの秘密 - CNET Japan
blog.keitap.com: Google's BigTable
今日の井原 - googleのJeff Dean氏がワシントン大学で行った講義をてきとーに解説してみる
Cafe Babe - Nutch on MapReduce
Google 技術講演会: MapReduce 〜大規模クラスタでの簡単なデータ処理 〜 2006年 3月14日(火) 18:40〜19:40

この中で僕の知らない新しい語が2つ出てきました．「Global Work Queue」と「BigTable」です．Sawzallも含めて読まねば．もう， GFS や PageRank を知っているくらいでは，Googleの技術を知っているとは言えませんね．

■2006/02/16のチェック
・×ビジョンの意識
・△ビジョンにそった行動
・△回りの人を思いやっているか
・○はてブに3以上ブクマ
・○blogを書く
・○専門分野の知識を得る
・○専門外の知識を得る
・×仕事とは別に何かを作り続ける
・○ピアノの練習
・○朝きちんと起きる