hadoop mapreduce執行過程是怎么樣的

# Hadoop MapReduce執行過程是怎么樣的

## 一、MapReduce概述

### 1.1 什么是MapReduce
MapReduce是一種分布式計算模型，由Google提出，主要用于大規模數據集（大于1TB）的并行運算。它將復雜的、運行于大規模集群上的并行計算過程高度抽象為兩個函數：Map和Reduce。

### 1.2 MapReduce的核心思想
"分而治之"是MapReduce的核心思想：
- **Map階段**：將大數據集分解為成百上千的小數據集
- **Reduce階段**：對Map階段的中間結果進行匯總

### 1.3 MapReduce的優勢
1. 易于編程
2. 良好的擴展性
3. 高容錯性
4. 適合海量數據離線處理

## 二、MapReduce架構組成

### 2.1 主要組件
| 組件 | 功能描述 |
|------|----------|
| Client | 提交MapReduce作業 |
| JobTracker | 資源管理和作業調度 |
| TaskTracker | 執行具體任務 |
| HDFS | 存儲輸入輸出數據 |

### 2.2 角色劃分
1. **JobTracker**（主節點）：
   - 管理所有作業
   - 調度任務到TaskTracker
   - 監控任務執行

2. **TaskTracker**（從節點）：
   - 執行Map和Reduce任務
   - 向JobTracker匯報狀態

## 三、MapReduce詳細執行流程

### 3.1 整體流程圖
```mermaid
graph TD
    A[Input Data] --> B[Split]
    B --> C[Map Task]
    C --> D[Shuffle]
    D --> E[Reduce Task]
    E --> F[Output]

3.2 階段分解

3.2.1 Input Split階段

1. 輸入文件被邏輯劃分為多個Split
2. 每個Split對應一個Map任務
3. 默認Split大小等于HDFS塊大?。?28MB）

關鍵參數：

<property>
    <name>mapreduce.input.fileinputformat.split.minsize</name>
    <value>1</value>
</property>

3.2.2 Map階段

1. 每個Map任務處理一個Split
2. 執行用戶定義的map()函數
3. 輸出形式的中間結果

示例代碼：

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
    
    public void map(LongWritable key, Text value, Context context) 
        throws IOException, InterruptedException {
        String[] words = value.toString().split(" ");
        for (String w : words) {
            word.set(w);
            context.write(word, one);
        }
    }
}

3.2.3 Shuffle階段（核心階段）

Map端Shuffle： 1. Partition：根據Reduce數量分區

   public int getPartition(K key, V value, int numReduceTasks) {
       return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
   }

1. Sort：按key排序
2. Spill：內存緩沖區溢出到磁盤
3. Merge：合并多個溢出文件

Reduce端Shuffle： 1. Copy：從各Map節點拉取數據 2. Merge：合并來自不同Map的數據 3. Sort：二次排序

3.2.4 Reduce階段

1. 執行用戶定義的reduce()函數
2. 對相同key的values集合進行處理
3. 輸出最終結果到HDFS

示例代碼：

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    public void reduce(Text key, Iterable<IntWritable> values, Context context) 
        throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

四、關鍵機制詳解

4.1 數據本地化優化

1. 原則：移動計算比移動數據更高效
2. 三級本地化策略：
   • 節點本地化（最佳）
   • 機架本地化
   • 跨機架訪問

4.2 容錯機制

1. TaskTracker故障：

   • JobTracker重新調度任務
   • 黑名單機制

2. Task失敗：

   • 自動重試（默認4次）
   • 超過重試次數則標記作業失敗

3. 推測執行：

   <property>
      <name>mapreduce.map.speculative</name>
      <value>true</value>
   </property>
   

4.3 性能優化技術

4.3.1 Combiner優化

job.setCombinerClass(WordCountReducer.class);

4.3.2 壓縮設置

<property>
    <name>mapreduce.map.output.compress</name>
    <value>true</value>
</property>

4.3.3 內存調優

<property>
    <name>mapreduce.map.memory.mb</name>
    <value>2048</value>
</property>

五、YARN架構下的MapReduce

5.1 組件變化

5.2 執行流程對比

1. 資源請求方式不同
2. 任務調度機制變化
3. 容錯實現方式更新

六、實際案例分析

6.1 單詞計數（WordCount）

完整代碼示例：

public class WordCount {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count");
        
        job.setJarByClass(WordCount.class);
        job.setMapperClass(WordCountMapper.class);
        job.setCombinerClass(WordCountReducer.class);
        job.setReducerClass(WordCountReducer.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

6.2 性能瓶頸分析

1. 數據傾斜解決方案：

   • 自定義Partitioner
   • 增加Reduce數量
   • 使用Combiner

2. Shuffle優化：

   • 調整緩沖區大小
   • 優化排序算法
   • 合理設置Reduce數量

七、MapReduce的局限與發展

7.1 主要局限性

1. 實時計算能力弱
2. 迭代計算效率低
3. 中間結果寫磁盤影響性能

7.2 替代技術

1. Spark：內存計算框架
2. Flink：流批一體處理
3. Tez：DAG執行引擎

八、最佳實踐建議

1. 輸入文件處理：

   • 盡量使用大文件（>128MB）
   • 避免大量小文件

2. 參數調優：

   <!-- 設置Map任務數 -->
   <property>
      <name>mapreduce.job.maps</name>
      <value>100</value>
   </property>
   

3. 監控工具：

   • JobHistory Server
   • YARN ResourceManager UI

結語

MapReduce作為Hadoop的核心計算框架，雖然在新一代計算框架面前顯得效率不足，但其”分而治之”的思想仍然深刻影響著大數據處理領域。理解MapReduce的執行原理，不僅有助于優化傳統Hadoop作業，更能為學習其他分布式計算框架奠定堅實基礎。 “`