MapReduce計算框架

2019/2/18 星期一

MapReduce計算框架
Mapreduce 是一個分布式的運算編程框架，核心功能是將用戶編寫的核心邏輯代碼分布式地
運行在一個集群的很多服務器上；

為什么要MAPREDUCE
（1）海量數據在單機上處理因為硬件資源限制，無法勝任，因為需要采用分布式集群的方式來處理。
（2）而一旦將單機版程序擴展到集群來分布式運行，將極大地增加程序的復雜度和開發難度
（3）引入mapreduce 框架后，開發人員可以將絕大部分工作集中在業務邏輯的開發上，而將分布式計算中的復雜×××由框架來處理

MAPREDUCE 程序運行演示
Hadoop 的發布包中內置了一個hadoop-mapreduce-example-2.4.1.jar，這個jar 包中有各種MR
示例程序，可以通過以下步驟運行：
啟動hdfs，yarn
然后在集群中的任意一臺服務器上執行，（比如運行wordcount）：
hadoop jar hadoop-mapreduce-example-2.4.1.jar wordcount /wordcount/data /wordcount/out

MapReduce引入的問題
1、分發程序，并啟動分發的程序
2、中間數據的緩存和調度
3、任務監控及失敗處理

MapReduce框架運行機制

MapReduce分為3個過程：
1、map //A讀取文件 B調用業務邏輯代碼（程序員只關系這個部分） C收集調用結果
2、shuffle機制 //緩存一下
3、reduce //A拉取緩存中的數據 B調用業務邏輯代碼（程序員只關系這個部分） C收集結果輸出（最終結果）默認把最終結果寫到hdfs中

MapReduce運行機制的數據流程

1、map //key：行起始偏移量 value：行的內容
2、shuffle //洗牌 按key分發：相同的key的kv必定會發給相同的reduce task
3、reduce //將key的值相同的整合成一組

mapreduce框架中的shuffle機制詳解
Shuffle 緩存流程：
----shuffle 是MR 處理流程中的一個過程，它的每一個處理步驟是分散在各個maptask 和reduce task 節點上完成的，整體來看，分為3 個操作：
1、分區partition
2、Sort 根據key 排序
3、Combiner 進行局部value 的合并

shuffle階段文字詳解
1、map階段先拿數據過來之后，會先調用map方法（我們自定義的）
2、拿到之后，map中會有一個context.write的輸出結果 map端的輸出結果就給到了shuffle階段了
3、在map端有一個環形緩沖區（默認內存大小100M）【實現的功能就是把這些kv收集起來】collect thread 收集線程
4、在不斷輸出，不斷收集的過程中環形的緩存區會不斷的寫，會寫滿，那么內部的機制是不會讓他寫滿，寫到80%就會溢出，還是在map端會把溢出來的數據被 （線程split thread）管理 在這里還會把溢出來的數據進行partition（分區） sort（排序）接下里split thread會把溢出來的數據存放到磁盤上面【這里存放在磁盤中的數據是分好區 排序好了的】。溢出文件分好區，且區內有序。
5、在map端，最后一次，會把數據全部的溢出來，也是分好區且區內有序的。然后會形成很多一系列分好區的小文件，接下來會進行merge（合并）小文件合并后形成大文件。這種合并是把分區內的數據一一對應的合并 所有1號區合并形成1號區 ... 這里同樣是分區且區內有序（這是最后在map端形成的最后文件形式）。
6、shuffle不是在某一個節點上完成的。shuffle是map和reduce中間的數據調度機制過程 主要包括：緩存 分區 排序
7、reduce 端 reduce主動下載map端的最后形成的文件（先主動下載所有map端的1號區的內容）。這里1號區 2號區 0號區會被分別在不同的reduce task中
8、接下來，會把從map端的1號區中都拿個過來的數據進行一次reduce端的merge（合并）并排序 //歸并排序
9、每個聚合調用一次reduce方法 傳遞的的參數 key：是這聚合組的相同的key，values：是這一聚合組的所有value的迭代器
//產生聚合values 迭代器來傳遞給reduce 方法，并把這組聚合kv（聚合的依據是GroupingComparator）中排序最前的kv 的key 傳給reduce 方法的入參key 。 最終會形成一個有序的且歸檔的文件
提示:其他的reduce也是做相同的事情，只不過其他的reduce拿到的數據可能是1號區 2號區的內容，處理的過程同上。每個reduce task會形成一個最終的有序結果文件
10、reduce端最后形成的文件，在內部有序，但是在全部不一定有序，這個需要我們程序去干預 如果是全局排序的話，需要加上分區的控制，讓這個分區按照一定的區段分區，最終形成reduce的全局有序。在某一個分界點 前面的一個key一個區，中間的一個key一個區，最后的key一個區等。

小結：整個shuffle 的大流程如下：
? map task 輸出結果到一個內存緩存，并溢出為磁盤文件
? combiner 調用
? 分區/排序
? reduce task 拉取map 輸出文件中對應的分區數據
? reduce 端歸并排序
產生聚合values 迭代器來傳遞給reduce 方法，并把這組聚合kv（聚合的依據是GroupingComparator）中排序最前的kv 的key 傳給reduce 方法的入參key

shuffle不是在某一個節點上完成的。shuffle是map和reduce中間的數據調度機制過程 主要包括：緩存 分區 排序

在 MapReduce中的整個過程中有6次是需要進行io操作的，分別為：
1、在第一次拿數據（從hdfs中拿數據到map中發生第一次io操作）
2、溢出數據(發生第二次io操作)
3、merge(小文件合并為大文件發生第三次io操作)
4、combiner局部合并(發生第四次io操作)
5、歸并排序(conbiner合并到reduce處理的過程發生第5次io操作)
6、reduce處理完的結果存儲在hdfs上發生第6次io操作
這也是MapReduce與spark對比的最大的瓶頸 spark只有 在第一次從hdfs上拿數據發生io操作，和處理完任務之后，把文件存儲在hdfs上也發生一次io操作，其中間的所有的處理過程都是在內存中，所有不存在 大量的io操作，速度快，所有spark為主流計算引擎。