MapReduce計算框架

2019/2/18 星期一

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

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

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

MapReduce引入的問題
1、分發(fā)程序，并啟動分發(fā)的程序
2、中間數(shù)據(jù)的緩存和調(diào)度
3、任務(wù)監(jiān)控及失敗處理

MapReduce框架運行機制

MapReduce分為3個過程：
1、map //A讀取文件 B調(diào)用業(yè)務(wù)邏輯代碼（程序員只關(guān)系這個部分） C收集調(diào)用結(jié)果
2、shuffle機制 //緩存一下
3、reduce //A拉取緩存中的數(shù)據(jù) B調(diào)用業(yè)務(wù)邏輯代碼（程序員只關(guān)系這個部分） C收集結(jié)果輸出（最終結(jié)果）默認(rèn)把最終結(jié)果寫到hdfs中

MapReduce運行機制的數(shù)據(jù)流程

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

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

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

小結(jié)：整個shuffle 的大流程如下：
? map task 輸出結(jié)果到一個內(nèi)存緩存，并溢出為磁盤文件
? combiner 調(diào)用
? 分區(qū)/排序
? reduce task 拉取map 輸出文件中對應(yīng)的分區(qū)數(shù)據(jù)
? reduce 端歸并排序
產(chǎn)生聚合values 迭代器來傳遞給reduce 方法，并把這組聚合kv（聚合的依據(jù)是GroupingComparator）中排序最前的kv 的key 傳給reduce 方法的入?yún)ey

shuffle不是在某一個節(jié)點上完成的。shuffle是map和reduce中間的數(shù)據(jù)調(diào)度機制過程 主要包括：緩存 分區(qū) 排序

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