MapReduce 核心流程深度解析：從 InputSplit 到 OutputCommitter 全鏈路

MapReduce 核心流程深度解析：從 InputSplit 到 OutputCommitter 全鏈路

MapReduce 是 Hadoop 計算引擎的核心編程模型。理解其內部流程——數據如何切分、Map 輸出如何排序溢寫、Shuffle 如何歸併、Reduce 如何分組輸出——對性能調優、故障排查、自定義組件開發至關重要。本文基於 Hadoop 2.x (YARN) 源碼，按數據流向串聯講解全鏈路關鍵類與機制。

整體三階段視角

1. Map：讀取分片 → 解析鍵值 → 用戶邏輯 → 輸出中間結果
2. Shuffle：Partition 分區 → 環形緩衝區排序/溢寫/合併/Combiner → Reduce 拉取/歸併排序
3. Reduce：分組迭代 → 用戶邏輯 → 寫最終結果

一、 YARN 資源調度與任務啟動

1.0 時代：JobTracker 單點調度

• Slot 模型（Map Slot / Reduce Slot）靜態劃分資源
• JT 同時負責資源管理、作業調度、Task 監控，擴展性瓶頸明顯

YARN 架構：關注點分離

• ResourceManager (RM)：全局資源調度，以 Container（CPU+Memory）為單位
• NodeManager (NM)：單節點資源監控、Container 生命週期管理
• ApplicationMaster (AM)：每個 Job 一個，向 RM 申請 Container、啟動 MapTask/ReduceTask、容錯重試

優勢：RM 僅管 Job 級資源，Task 級調度下放給 AM，支持多框架混部（Spark、Flink 等）。

二、 Map 階段：輸入切分與讀取

InputFormat → InputSplit → RecordReader

// 用戶配置
job.setInputFormatClass(TextInputFormat.class);

// 核心接口
public abstract class InputFormat {
    List getSplits(JobContext ctx);           // 邏輯切分
    RecordReader createRecordReader(InputSplit split, TaskAttemptContext ctx); // 物理讀取
}

• InputSplit：邏輯分片，僅含 length、locations（數據所在節點）。Split 數 = MapTask 數。
• RecordReader：將字節流轉為 <K, V> 記錄流。TextInputFormat 默認 LineRecordReader，按 \n 切行，Key=偏移量，Value=行文本。

MapTask 啟動流程

1. AM 向 RM 申請 Container
2. Container 啟動 YarnChild → MapTask.run()
3. MapTask.runNewMapper() 反射實例化 Mapper、InputFormat、RecordReader
4. 構建 MapContextImpl 注入 InputSplit、RecordReader、OutputCollector
5. 循環 context.nextKeyValue() → mapper.map()

三、 Map 輸出：環形緩衝區、溢寫、排序、Combiner

MapOutputCollector (默認 MapOutputBuffer)

Map 輸出不直接寫磁盤，先進入環形內存緩衝區（默認 100MB，mapreduce.task.io.sort.mb）。

關鍵數據結構

• kvbuffer：byte[] 存序列化後的 Key、Value、Partition
• kvmeta：int[] 索引元數據（每條 16 字節：partition、key offset、key length、value length）
• soft limit：緩衝區使用達 80%（mapreduce.map.sort.spill.percent）觸發溢寫

溢寫流程

1. 啟動獨立線程 SpillThread，不阻塞 Map 輸出
2. 對 kvmeta 按 (partition, key) 快速排序
3. 若配置 Combiner，對同 Key 值做局部聚合
4. 寫入磁盤文件 spillN.out，並生成索引 spillN.out.index

合併階段

Map 結束前，將所有溢寫文件合併為單一 mapoutput/file.out（若溢寫文件數 ≥ mapreduce.task.io.sort.factor 默認 10，採用多輪歸併）。最終輸出有序、分區、可選 Combiner 聚合後的文件。

四、 Shuffle：Reduce 拉取與歸併排序

ReduceTask 三階段

1. Copy (Copy Phase)：Fetcher 線程（默認 5 個）HTTP 拉取 Map 輸出。根據數據大小選擇 InMemoryMapOutput 或 OnDiskMapOutput。
2. Sort (Merge Phase)：拉取完成的數據進入 MergeManager，內存/磁盤歸併排序，產生 RawKeyValueIterator。
3. Reduce (Reduce Phase)：遍歷迭代器調用 reduce()。

Partitioner — 決定數據去向

public interface Partitioner {
    int getPartition(K key, V value, int numPartitions); // numPartitions = reduce task 數
}

• 默認 HashPartitioner：key.hashCode() & 0x7FFFFFFF % numPartitions
• 自定義 Partitioner 需保證分布均勻，否則導致 Data Skew（少數 Reduce 任務極慢）

GroupingComparator — 決定分組邊界

• Shuffle 後數據已按 (partition, key) 全局有序
• GroupingComparator 僅比較 Key，相鄰 Key 相等則歸為同一組，一次 reduce() 處理
• 默認 WritableComparator 按字節比較；自定義可實現「只比 ID 不比時間」等業務分組

五、 Reduce 階段：分組迭代與輸出

ReduceContext 核心邏輯

public void run(Context context) {
    setup(context);
    while (context.nextKey()) {                    // 移動到下一組
        reduce(context.getCurrentKey(),              // 當前 Key
               context.getValues(),                  // 同組 Value 迭代器
               context);
        // 重置備用存儲（如有）
    }
    cleanup(context);
}

• getValues() 返回 Iterable，底層封裝 RawKeyValueIterator，按需讀取，支持超大 Value 流式處理。

六、 輸出提交：OutputFormat、RecordWriter、OutputCommitter

OutputFormat 雙重職責

1. checkOutputSpecs()：驗證輸出目錄不存在（防誤覆蓋），拋 FileAlreadyExistsException
2. getRecordWriter()：返回 RecordWriter 寫數據
3. getOutputCommitter()：返回 OutputCommitter 管理兩階段提交

FileOutputCommitter 兩階段提交

目錄	用途
${output}/_temporary/${attemptId}	Task 臨時輸出目錄
${output}/_temporary	Job 級臨時目錄（所有成功 Task 移至此）
${output}	最終輸出目錄

1. Task 寫入 _temporary/${attemptId}
2. Task 成功 → AM 調用 commitTask() 移至 _temporary
3. 所有 Task 成功 → AM 調用 commitJob() 重命名 _temporary → ${output}，生成 _SUCCESS 標記
4. Task 失敗 → abortTask() 刪除臨時目錄，重試時乾淨無殘留

RecordWriter 寫入細節

• TextOutputFormat 默認 LineRecordWriter，Key/Value 以 \t 分隔，換行結束
• 支持壓縮（mapreduce.output.fileoutputformat.compress=true + CompressionCodec）
• 文件命名：part-r-00000，FileOutputFormat.getUniqueFile() 保證唯一性（含 Task 類型 m/r、分區號）

七、 實戰調優清單

症狀	排查方向	典型參數/手段
Map 端溢寫頻繁	緩衝區過小	增大 mapreduce.task.io.sort.mb、調低 spill.percent
Reduce 端 Data Skew	Partition 分布不均	自定義 Partitioner、Key 加鹽、Combine 輸入預聚合
Shuffle 讀取超時	網絡/磁盤慢	增大 mapreduce.reduce.shuffle.parallelcopies、啟用壓縮 mapreduce.map.output.compress
小文件過多	Map 數過大/Reduce 數過小	調整 mapreduce.input.fileinputformat.split.maxsize、合併小文件
Combiner 無效	邏輯不滿足結合律	僅用於可交換/結合操作（求和、計數、最大值），去重、平均值不可用

八、 總結

MapReduce 全鏈路核心在於 分治 + 排序 + 兩階段提交：

1. InputSplit 決定並行度；RecordReader 決定記錄邊界
2. Map 端 環形緩衝區 + 快速排序 + Combiner 最小化磁盤 I/O
3. Partitioner 決定數據去向；GroupingComparator 決定分組粒度
4. Reduce 端 歸併排序 + 流式迭代 支持超大數據組
5. OutputCommitter 保證輸出原子性、可重試

掌握這些機制，不僅能寫出正確的 MapReduce 程序，更能在面對數據傾斜、OOM、小文件、任務卡死等生產疑難雜症時，從源碼層面精準定位、精準施策。