MR編程模型及MR V1講解

MR編程模型

MR編程模型主要分為五個步驟：輸入、映射、分組、規約、輸出。

1. 輸入（InputFormat）：
   主要包含兩個步驟—數據分片、迭代輸入

   數據分片（getSplits）：數據分為多少個splits，就有多少個map task; 
   單個split的大小，由設置的split.minsize和split.maxsize決定;
   公式為 max{minsize, min{maxsize, blocksize}}； 
   hadoop2.7.3之前blocksize默認64M，之后默認128M。
   
   決定了單個split大小之后，就是hosts選擇，一個split可能包含多個block（將minsize設置大于128M）；
   而多個block可能分布在多個hosts節點上（一個block默認3備份，如果4個block就可能在12個節點），getsplits會選擇包含數據最多的一部分hosts。
   由此可見，為了讓數據本地話更合理，最好是一個block一個task，也就是說split大小跟block大小一致。
   
   getSplits會產生兩個文件
   job.split：存儲的主要是每個分片對應的HDFS文件路徑，和其在HDFS文件中的起始位置、長度等信息（map task使用，獲取分片的具體位置）；
   job.splitmetainfo：存儲的則是每個分片在分片數據文件job.split中的起始位置、分片大小和hosts等信息（主要是作業初始化時使用，用于map task的本地化）。
   
   迭代輸入：迭代輸入一條條的數據，對于文本數據來說，key就是行號、value當前行文本。

2. 映射（map）：正常的map操作，將一對kv映射成為另外一對kv
3. 分組（partition）：
   按照設置的reduce個數來進行分組，getPartitions共三個參數：k、v、partitionnum；
   默認按照HashPartition，如果需要全排序，也可以設置TotalOrderPartitioner，它會采樣一部分數據排序后設置R-1（R是reduce個數）個分割點，保證map task生成的R個文件的文件與文件之間的數據都是有序的，reduce只需要對單個文件內部再排序即可。
4. 規約（reduce）：reduce做聚合處理。
5. 輸出（OutputFormat）：
   一件事情是檢查輸出目錄是否存在，如果存在則報錯；
   另一件事情是將數據輸出到臨時目錄。

作業提交及初始化

1. 作業提交與初始化大概分為4個步驟：執行提交、client上傳文件到hdfs、client與JobTracker通信提交任務、JobTracker通知TaskScheduler初始化任務。
2. JobClient與JobTracker的通信過程如下兩所示
3. 作業提交時序圖
   第一步：JobClient先跟JobTracker交互獲取到一個jobid；
   第二步：JobClient與HDFS交互創建輸出目錄；
   第三步：與HDFS交互上傳任務運行所以來的文件（配置文件、jar包等）
   第四步：JobClient調用getSplits，與HDFS交互生成分片信息并寫到分片文件中；
   第五步：與jobtracker交互提交任務。
4. JobTracker收到任務提交請求后會先生成一個JobInProgress對象，這個對象會管理和監控這個job的整個運行狀況；之后JobTracker再告訴TaskSchduler進行作業初始化。
5. 作業初始話大致過程如下

JobTracker與TaskTracker

1. JobTracker主要負責作業的運行時管理，以三級樹的方式進行管理：首先會給作業初始化一個對象JobInProgress，初始化后每個task有個TaskInProgress，每個task對應多個TaskAtempt。其中一個TA成功則此TI成功，所有TI成功則此job成功
   JobTracker將很多數據以KV形式存儲在map中，比如jobs存儲的是jobid和JobInProgress的映射；
   JobTracker通過接收TaskTracker的心跳請求，并發出應答來監控和管理作業運行過程，在應答中會下達各種命令：運行新task、殺死task等等
2. TaskTracker：在每臺機器上會啟動一個TaskTracker進程，不斷地向JobTracker發送心跳，匯報當前節點的資源使用情況、當前節點的task運行情況，并根據JobTracker在應答中的指令執行具體命令
   TaskTracker會為每個task啟動一個JVM（可重用，但是僅限于重用同類型任務）
   TaskTracker啟動一個新任務
   第一步：先進行作業本地化，某個作業在TaskTracker上的第一個task會進行作業本地化，也就是把作業運行依賴的文件、jar包從hdfs下載到本地。（為避免多個task同時進行作業本地化，會對本地化操作加鎖）；
   第二步：創建任務臨時目錄；
   第三步：啟動JVM，并在JVM運行任務（部分情況JVM可復用）；

Map Task內部運行過程

map task總共可以五個過程：read、map、collect、splill、conbine。
Read：從數據源讀入一條條數據；
map：將數據傳給map函數，變成另外一對KV
collect階段：
主要是map處理完的數據，先放入內存的環形緩沖區中，待環形緩沖區的值超過一定比例的時候再執行下一步的spill到磁盤；
collect()內部會調用getPartition來進行分區，而環形緩沖區則存儲的是K、V和partition號
這里采用的兩級索引結構，主要是排序時在同一個partition內排序，所以先排partition，再排partition內部數據。
kvindices中記錄的分區號、key開始的位置、value開始的位置，也就是一對兒KV在kvindices中占用3個int，kvoffsets只記錄一對KV在kvindices中的偏移地址，所以只需要一個int，所以二者按1：3的大小分配內存。
spill過程：
環形緩存區中內存數據在超過一定閾值后會spill到磁盤上，在splill到磁盤上之前會先在內存中進行排序（快速排序）；
之后按分區編號分別寫到臨時文件，同一個分區編號后面會有個數字，表示第幾次溢寫，conbine：對多個文件合并，多倫遞歸，沒輪合并最小的n個文件。

Reduce Task內部運行過程

reduce總共可分為以下幾個階段：shuffle、merge、sort、reduce、write
shuffle：從JobTracker中獲取已完成的map task列表以及輸出位置，通過http接口獲取數據；
merge：shuffle拉去的數據線放入內存，內存不夠再放入磁盤，會有一個線程不斷地合并內存和磁盤中的數據
sort：reduce從不同的map task中拉取到多個有序文件，然后再做一次歸并排序，則每個reduce獲取到文件就都是有序的了