公開2024.09.20更新2026.10.28 19 人が閲覧（直近 30 日）データ分析・BI

MapReduceとは？仕組みとPythonで動かすワードカウント・Hadoopとの違い

MapReduceは、Googleが2004年に発表した、大量のデータを多数のサーバーに分けて並列に集計するためのプログラミングモデルです。開発者は「1件ずつ整形してキーと値を出す処理（Map）」と「同じキーの値をまとめる処理（Reduce）」の2つだけを書き、データの分割・配送・故障したマシンの処理のやり直しは実行基盤が引き受けます。この記事では仕組みを3段階で説明したうえで、Pythonで書いたワードカウントをHadoop 3.5.0で実際に動かした出力と、Combinerを足したときにシャッフル量がどれだけ減るかを示します。後半はHadoopやApache Sparkとの違いと、2026年の今MapReduceを選ぶ場面をまとめます。

まとめ

定義：大規模データを分割して並列処理し、キーごとに集約する分散バッチ処理のモデル。原典はJeffrey DeanとSanjay GhemawatのOSDI 2004論文。
仕組み：Map → Shuffle&Sort → Reduce の3段階。同じキーを1台のReducerへ集めるShuffleが、ネットワーク負荷と性能を左右する。
動かし方：Hadoop Streamingを使えばPythonのスクリプトをそのままmapperとreducerにできる。Hadoop 3.5.0はJava 17で動く。
Combiner：Map側で先に足し込むと、5万件のワードカウントでシャッフル量が530,006バイトから50バイトに減った。合計や最大値はそのまま使え、平均は（合計, 件数）の組に直してから使う。
Hadoopとの違い：MapReduceは処理モデル、Hadoopはそれを動かすソフトウェア（Hadoop Common・HDFS・YARN・MapReduce）。
今使うか：新規の分析・機械学習・ストリーム処理はSparkやFlinkを選ぶ。MapReduceは既存Hadoop資産の保守と、仕組みを学ぶ教材としての価値が中心。

MapReduceとは：Googleが2004年に発表した分散処理のモデル

MapReduceの原典は、GoogleのJeffrey DeanとSanjay Ghemawatが2004年12月のOSDI（6th Symposium on Operating Systems Design and Implementation）で発表した論文MapReduce: Simplified Data Processing on Large Clustersです。要旨には、当時すでにGoogleのクラスタで1日1,000件を超えるMapReduceジョブが動いていたと書かれています。名前は関数型プログラミングのmap（各要素に同じ関数を当てる）とreduce（要素を畳み込んで1つにする）に由来します。

この論文が示したのは、並列化・負荷分散・故障への対処という分散処理の面倒な部分をライブラリ側に閉じ込め、利用者にはMapとReduceの2つの関数だけを書かせる、という分担です。論文は考え方と社内実装を説明したもので、ソフトウェアとしては公開されていません。誰でも使える実装として広まったのは、後に出てくるApache Hadoopです。

分散処理とは：1台で終わらない処理を複数台に分ける方式

分散処理とは、1つの処理を複数のコンピュータに分けて同時に実行する方式です。データがテラバイト級になり、ディスクの読み出し速度が処理時間を決めるようになると、CPUやメモリを増やすスケールアップでは読み込みの時間がほとんど縮まりません。台数を増やす（スケールアウト）と速くはなりますが、どのマシンにどのデータを渡すか、途中で1台壊れたらどうするかを利用者が全部書くのは現実的ではありません。MapReduceは、処理を「行ごとに独立した変換」と「キーごとの集約」の型にはめることで、この割り振りを基盤側で自動化しました。論文の評価環境は約1,800台のクラスタで、約1テラバイトのデータの検索と並べ替えを測っています。分散処理の基盤はその後、Pythonの関数をそのまま並列化するRayのように、MapReduceの型にはまらない処理を扱う方向へ広がっています。

MapReduceの仕組み：Map・Shuffle・Reduceの3段階

MapReduceは「MapとReduceの2段階」と説明されることが多いのですが、実際の処理は間にShuffle&Sortを挟んだ3段階です。apple banana apple という1行を入力にすると、データは次のように形を変えます。

入力        : apple banana apple

Map出力     : (apple, 1) (banana, 1) (apple, 1)
              単語をキー、1を値にしたペアを行ごとに出す

Shuffle後   : (apple, [1, 1]) (banana, [1])
              同じキーの値を1か所に集め、キー順に並べる

Reduce出力  : (apple, 2) (banana, 1)
              キーごとに値を合計する

Map：入力を一定の大きさの断片（スプリット）に分け、断片ごとに並列でキーと値のペアを出力する。行どうしは独立しているので、スプリットの数までは台数を増やして並列に処理できる。
Shuffle&Sort：全Mapの出力を、同じキーが同じReducerに届くように振り分けて転送し、キー順に並べる。全マシン間でデータが行き交うので、ここがネットワークの混む工程になる。
Reduce：キーごとにまとまった値を受け取り、合計・件数・最大値などの最終結果を出す。

集計の多くはこの型に収まります。Reduceで合計を件数に替えればアクセスログのURL別件数、最大値に替えればユーザーごとの最終アクセス日時になります。

Partitionerによるキーの振り分けと偏り

Shuffleで「同じキーは同じReducerへ」を保証しているのがPartitionerです。原論文では、中間キーの空間をhash(key) mod R（Rは指定したReduceタスク数）でR個に分ける方法を既定の例として挙げています。キーのハッシュ値で振り分けるため、特定のキーに値が集中すると、そのキーを受け持つReducerだけが遅れて全体の完了が待たされます。アクセスログでトップページのURLだけが極端に多いといった偏りがある場合は、キーに接尾辞を付けて複数のReducerへ分けて部分集計し、もう一段のジョブで接尾辞を外して元のキーごとに合算する、2段階の集計が必要になります。

故障に強い理由：再実行・データローカリティ・バックアップタスク

MapReduceが数千台規模で実用になったのは、故障を前提にした設計だからです。原論文が挙げる仕組みは次の3つです。

再実行：故障したワーカーで実行中だったMapタスクとReduceタスクは、別のマシンで最初からやり直す。完了済みのMapタスクも、出力が故障したマシンのローカルディスクにあって読めなくなるため再実行する。完了済みのReduceタスクは、出力が共有のファイルシステムにあるので再実行しない。
データローカリティ：入力はGFS上で64MBのブロックに分かれ、通常3つの複製が別々のマシンにある。マスターはデータの置き場所を見て、そのデータを持つマシン（無理なら近いマシン）にMapタスクを割り当て、ネットワーク転送を減らす。
バックアップタスク：全体の完了間際に、遅れているタスク（ストラグラー）の予備実行を別マシンで走らせ、先に終わった方を採用する。論文の並べ替えの例では、この仕組みを止めると完了まで44%長くかかった。

原論文は、Map関数とReduce関数が入力だけで結果の決まる決定的な処理であれば、再実行が起きても故障のない逐次実行と同じ出力になると説明しています。乱数や現在時刻に左右される非決定的な処理では、この保証が弱まります。外部への書き込みは別の問題で、論文は副作用を原子的かつ冪等にすることを利用者の責任としています。Map関数の中でデータベースに1件ずつ書き込むと、再実行やバックアップタスクのたびに同じ行が二重に書き込まれます。

PythonによるMapReduceのワードカウント実行手順

Googleの原実装はC++のライブラリで、それを手本にしたApache HadoopのMapReduceはJavaで書かれています。Javaを書かなくても、Hadoopに同梱されているHadoop Streamingを使うと、標準入力を読み標準出力に書く任意の実行ファイルをmapperとreducerにできます。ここではPythonで書いたワードカウントを、まずHadoopなしで、次にHadoop 3.5.0で動かします。

mapper.pyとreducer.pyの実装

mapperの仕事は、1行を単語に分けて単語<タブ>1を1行ずつ出力することだけです。

#!/usr/bin/env python3
import sys

for line in sys.stdin:
    for word in line.split():
        print(f"{word}\t1")

reducerは、キー順に並んだ入力を前提に、キーが変わった時点で前のキーの合計を出力します。Hadoop StreamingはReducerに「キーごとの値のリスト」ではなく、並べ替え済みの行をそのまま流すので、キーの切れ目は自分で判定します。

#!/usr/bin/env python3
import sys

current, count = None, 0
for line in sys.stdin:
    word, n = line.rstrip("\n").split("\t")
    if word != current:
        if current is not None:
            print(f"{current}\t{count}")
        current, count = word, 0
    count += int(n)
if current is not None:
    print(f"{current}\t{count}")

Hadoopなしの動作確認：sortによる並べ替えの代用

1台で試すだけなら、Shuffle&Sortのうち並べ替え（同じキーを隣に並べる）の部分はUnixのsortで代用できます。複数のマシンへの振り分けと転送はこの例には含まれません。入力を2行用意して、Map→Sort→Reduceをパイプでつなぎます。

$ printf 'apple banana apple\nbanana cherry\n' > input.txt
$ python3 mapper.py < input.txt | LC_ALL=C sort | python3 reducer.py
apple	2
banana	2
cherry	1

sortを抜くとappleが離れた位置に2回現れ、reducerが同じ単語を2回に分けて出力します。Shuffleがなぜ必要かは、この1行を外すとよく分かります。

Hadoop 3.5.0のローカルモードでのHadoop Streaming実行

2026年10月時点のApache Hadoopの最新版は、2026年4月2日に出た3.5.0です。3.5系はJava 17に正式対応した最初のリリースで、サーバー側はJava 17が必須、クライアント側はJava 17と21に対応します。3.4系（最新3.4.3）はJava 8と11で動くので、手元のJavaに合わせて選びます。

配布物を展開してJAVA_HOMEをJava 17に向ければ、HDFSもYARNも起動しないローカルモードでジョブが動きます。入力はローカルのディレクトリ、出力先は存在しないディレクトリを指定します。

$ export JAVA_HOME=<Java 17のパス>
$ export HADOOP_HOME=<展開先>/hadoop-3.5.0
$ mkdir in && cp input.txt in/
$ $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-3.5.0.jar \
    -files mapper.py,reducer.py \
    -mapper "python3 mapper.py" \
    -reducer "python3 reducer.py" \
    -input in -output out
...
INFO mapreduce.Job: Job job_local1458696534_0001 completed successfully
        Map input records=2
        Map output records=5
        Reduce input groups=3
        Reduce shuffle bytes=59
        Reduce output records=3
$ cat out/part-00000
apple	2
banana	2
cherry	1

ジョブ終了時に表示されるカウンタが、3段階の中身をそのまま表しています。入力2行（Map input records）から5つのペア（Map output records）が出て、Shuffleで3つのキー（Reduce input groups）にまとまり、3行の結果（Reduce output records）になりました。出力先に既存のディレクトリを指定するとジョブは開始前にエラーで止まるので、再実行するときはoutを消すか別名にします。実際のクラスタでは-inputと-outputにHDFSやS3のパスを渡し、同じコマンドをAmazon EMRやCloud Dataprocのクラスタ上で実行します。

Combinerによるシャッフル量の削減：530,006バイトから50バイトへ

ワードカウントのMap出力は(the, 1)のような同じペアの繰り返しで、そのままではネットワーク越しに大量の「1」を送ることになります。原論文はこれを減らす仕組みとして、Map側で部分的に集約してから送るCombiner関数を挙げ、多くの場合Reduceと同じコードを使えるとしています。Hadoop Streamingでは-combinerオプションにreducerと同じスクリプトを渡すだけです。

先ほどの2行を1万回繰り返した2万行（5万語）を入力に、前節のコマンドへ-combiner "python3 reducer.py"を足した場合と足さない場合でカウンタを比べました。

カウンタ	Combinerなし	Combinerあり
Map output records	50,000	50,000
Combine output records	0	3
Reduce shuffle bytes	530,006	50
Reduce input records	50,000	3
最終結果	apple 20000 ほか2行（計3行）	同じ

結果は同じで、Reducerが受け取るデータは5万件から3件に減りました。今回は入力ファイルが小さく、スプリット（Mapタスク）が1つだったので、全データが1回で畳み込まれた最良のケースです。実際のクラスタではMapタスクごとに部分集約されるため、減り方はキーの重複の多さに左右されます。

Hadoop公式のMapReduceチュートリアルのJava版ワードカウントも、同じIntSumReducerをCombinerとReducerの両方に設定しています。ただしCombinerは常に使えるわけではありません。同じチュートリアルが、ディスクへの書き出し（spill）の途中でCombinerを通るかどうかは未定義と書いているように、適用される回数もタイミングも保証されないので、部分集計の結果をもう一度入力しても最終結果が変わらない処理、つまり合計や最大値のように結合則と交換則が成り立つ処理に限られます。件数を数える場合も、Combinerの出力は（キー, 部分件数）になるので、Reducerではレコード数を数え直さずに部分件数を足し合わせます。平均値をそのままCombinerにすると「平均の平均」になって結果が狂うため、Mapで(合計, 件数)の組を出し、Reduceで最後に割る形に直してから使います。

MapReduceとHadoopの違い

MapReduceは処理の考え方（プログラミングモデル）で、Hadoopはそれを実行するオープンソースのソフトウェア群です。Googleの論文の考え方を実装したのがApache Hadoopで、公式のモジュール一覧は共通ライブラリのHadoop Commonに加えて次の3つを挙げています。

HDFS：ファイルをブロックに分け、複製して多数のマシンに置く分散ファイルシステム。GFSに相当する。
YARN：クラスタのCPUとメモリをジョブに割り当てるリソース管理。Hadoop 2から、MapReduce以外の処理エンジンもこの上で動くようになった。
Hadoop MapReduce：YARNの上でMapReduceジョブを実行するエンジン。上の実行例で使ったのはこれ。

つまり「HadoopでMapReduceを動かす」が正しい関係で、HadoopクラスタではMapReduceの代わりにSparkやHiveのジョブを動かすこともできます。クラウドではこのHadoop環境をマネージドで借りるのが一般的で、AWSのAmazon EMRは以前Amazon Elastic MapReduceという名前でした。

MapReduceとApache Sparkの違いと今の採用判断

結論から言うと、新しくデータ処理を作るときにMapReduceを第一候補にする場面はほぼありません。ただしHadoopのMapReduceエンジン自体は3.5.0でも同梱され、保守が続いています。

SparkとFlinkに主役が移った理由

差が出るのは、同じデータを何度も読む処理です。MapReduceはジョブごとに中間結果をディスクへ書き、次のジョブがそれを読み直します。機械学習のように同じデータで計算を何十回も繰り返すと、この読み書きが処理時間の大半を占めます。Sparkは中間データをメモリに保持できるRDDという仕組みでこれを避けました。Sparkの基になったRDDの論文（NSDI 2012）は、反復する機械学習やグラフの処理でHadoopより最大20倍速いと報告しています。よく引用される「最大100倍」は2013年の講演資料の表現で、2026年10月時点のSpark公式トップページにはありません。公式FAQが載せている比較は、2014年のソートベンチマークで100TBをHadoop MapReduceの10分の1の台数で3倍速く並べ替えたという事例です。

Hadoopの周辺でも置き換えが続いています。SQLでHadoop上のデータを扱うApache Hiveは、設定項目の説明で、MapReduce実行エンジン（hive.execution.engine=mr）をHive 2.0.0で非推奨にしました。当時の切り替え先はTezかSparkでしたが、Hive 4.0の変更点ではHive on Sparkが削除され、MapReduceエンジンは非推奨のまま残っています。Google自身も2014年6月のGoogle I/OでCloud Dataflowを先行公開し、発表ブログで、DataflowはMapReduceから発展したFlumeやMillWheelといった社内の後継技術のモデルに基づくと説明しています。現在のDataflowはApache Beamで書いたパイプラインを実行するサービスになっています（詳しくはCloud Dataflowの記事）。ストリーム処理を新しく組むなら、Apache Flinkが代表的な候補です。

MapReduceを選ぶ場面・避けるべき場面

MapReduceを選んでよいのは、次の条件に当てはまるときです。

Hadoop MapReduceで書かれたジョブが既に安定して動いており、書き換えのコストに見合う速度向上が要らない。
HDFSとYARNのクラスタが既にあり、単純な集計バッチを1本足すだけで、Sparkを導入して運用する体制を新たに用意する費用が見合わない。
分散処理の仕組み（分割・Shuffle・再実行）を学ぶ教材として、最小の構成で動かしたい。

逆に、反復計算を伴う機械学習、対話的な分析クエリ、秒単位の応答が要るストリーム処理を新しく作るなら、MapReduceは避けてください。反復計算はSparkのメモリ保持でディスクの読み書きを減らせ、対話的な分析はSpark SQLやTrino、秒単位の応答はFlinkの領域です。MapReduceはどの用途でもジョブごとのディスク読み書きが前提になります。既存のMapReduceジョブを移行するときは、集計ロジックをSpark SQLやPySparkに書き直し、ジョブ間の依存関係をApache Airflowのようなワークフロー基盤で管理する構成が一般的です。バッチとリアルタイムの両方を残す設計ならLambdaアーキテクチャも検討対象になります。issohではデータ分析基盤構築・MLOps構築支援として、Hadoop資産の棚卸しから移行先のデータ基盤づくりまでを支援しています。

Hadoop以外の「MapReduce」：MongoDBとJavaScript

MapReduceという言葉は、Hadoopの外でも使われます。検索で別の文脈に行き当たったときの区別のために、主なものを挙げます。

MongoDBのmapReduce：JavaScriptのMap関数とReduce関数で集計するデータベースのコマンド。MongoDB 5.0で非推奨になり、性能と使いやすさで勝る集計パイプライン（aggregation pipeline）への書き換えが案内されている。
JavaScriptのmapとreduce：配列のメソッドで、words.map(w => [w, 1])で変換しreduceで畳み込む。考え方は同じだが、1つのプロセスの中で動くだけで分散はしない。
SQLのGROUP BY：SELECT word, COUNT(*) FROM words GROUP BY wordはワードカウントと同じ集計。HiveやSpark SQLは、このSQLを分散処理の計画に変換して実行する。