Hadoop MapReduce 源碼深度分析：JobTracker 初始化 Job、TaskTracker 心跳回應、TaskScheduler 任務分配全流程

背景

基於 Hadoop 1.x (MRv1) 架構，分析 JobTracker 與 TaskTracker 交互核心流程：Job 提交 → 初始化 → TaskTracker 心跳 → 任務調度分配。

核心組件

• JobTracker：作業調度中心，管理 Job、Task、TaskTracker 狀態。
• TaskTracker：節點代理，執行 Task、周期性發送心跳。
• JobInProgress (JIP)：JobTracker 端作業狀態封裝。
• TaskInProgress (TIP)：TaskTracker 端任務執行狀態封裝。
• TaskScheduler：預設 JobQueueTaskScheduler，實現任務分配邏輯。

流程 1：Job 初始化 (EagerTaskInitializationListener)

1. 用戶提交 Job，構建 JobInProgress 加入 jobs 集合。
2. JobQueueJobInProgressListener 將 JIP 放入 jobQueue (Map<JobSchedulingInfo, JIP>)。
3. EagerTaskInitializationListener 將 JIP 放入 jobInitQueue，按優先級/提交時間排序，喚醒 JobInitManager 線程。
4. JobInitManager 從佇列取出 JIP，提交 InitJob 至線程池執行 JIP.initTasks()。

initTasks() 關鍵工作

• 讀取 InputSplit，驗證數量與 numMapTasks 一致。
• 建立 Map Task 的 TaskInProgress 陣列，放入 nonRunningMapCache (Map<Node, List<TIP>>) 實現資料本地性。
• 建立 Reduce Task TIP 陣列，放入 nonRunningReduces。
• 計算 completedMapsForReduceSlowstart（預設 5% Map 完成才啟動 Reduce）。
• 建立 Setup/Cleanup Task（各一 Map/Reduce）。
• 標記 tasksInited = true，記錄 JobHistory。

流程 2：TaskTracker 心跳處理 (JobTracker.heartbeat)

參數：TaskTrackerStatus、restarted、initialContact、acceptNewTasks、responseId。

主要邏輯

1. 驗證 Tracker 合法性 (acceptTaskTracker，檢查 include/exclude 清單)。
2. 處理重啟/初次聯繫：重啟標記健康、初次聯繫無記錄則要求 Reinit。
3. 檢查重複心跳：responseId 未變則重發上次回應。
4. processHeartbeat 更新 Tracker 狀態、Task 狀態、節點健康度。
5. 構建回應 HeartbeatResponse：

任務分配 (taskScheduler.assignTasks)

public List<Task> assignTasks(TaskTracker tracker)
{
    // 1. 計算集群/Tracker 級 Map/Reduce 容量與負載因子
    // 2. 分配 Map Task：優先 Node-local → Rack-local → Off-switch
    //    從 failedMaps、nonRunningMapCache、nonLocalMaps、speculative 依序嘗試
    // 3. 分配 Reduce Task：每次心跳最多 1 個，檢查 Map 完成比例
    // 4. 預留槽位 (exceededPadding) 給高優先級/失敗重試任務
    // 5. 返回 Task 列表 (MapTask/ReduceTask 實例)
}

Map Task 選取優先序

1. 失敗重試：failedMaps 中找未在該 Tracker 執行過的 TIP。
2. Node-local：nonRunningMapCache[trackerNode]。
3. Rack-local：向上遍歷拓撲樹 parent Node。
4. Off-switch：更高層級 Node（受 maxCacheLevel 限制）。
5. Non-local：nonLocalMaps (計算密集型，無輸入資料)。
6. 推測執行：runningMapCache 中緩慢的 TIP。

Reduce Task 選取

1. 檢查 Map 完成比例達標。
2. 從 nonRunningReduces 取一個。
3. 推測執行：runningReduces 中緩慢的。

流程 3：回應組裝與發送

• 將分配的 Task 封裝為 LaunchTaskAction 加入 actions。
• 加入 KillTaskAction、KillJobAction、CommitTaskAction（輸出提交）。
• 計算下次心跳間隔，更新 trackerToHeartbeatResponseMap。
• 清理標記任務 removeMarkedTasks。

JobQueueTaskScheduler 設計原則總結

1. 優先級高的 Job 先調度，同優先級 FIFO。
2. 追求 Tracker 級負載均衡（任務數量層面）。
3. 盡量分配本地任務，但限制非本地任務比例，保留並發度。
4. 預留槽位應對失敗重試與高優先級 Job。

源碼關鍵類參考

• org.apache.hadoop.mapred.JobTracker
• org.apache.hadoop.mapred.TaskTracker
• org.apache.hadoop.mapred.JobInProgress
• org.apache.hadoop.mapred.TaskInProgress
• org.apache.hadoop.mapred.JobQueueTaskScheduler
• org.apache.hadoop.mapred.EagerTaskInitializationListener