2019獨角獸企業(yè)重金招聘Python工程師標(biāo)準(zhǔn)>>>

1、MapReduce中數(shù)據(jù)流動
?? （1）最簡單的過程：? map - reduce
?? （2）定制了partitioner以將map的結(jié)果送往指定reducer的過程：　map - partition - reduce
?? （3）增加了在本地先進(jìn)性一次reduce（優(yōu)化）過程：　map - combin(本地reduce) - partition -reduce
2、Mapreduce中Partition的概念以及使用。
（1）Partition的原理和作用
??????? 得到map給的記錄后，他們該分配給哪些reducer來處理呢？hadoop采用的默認(rèn)的派發(fā)方式是根據(jù)散列值來派發(fā)的，但是實際中，這并不能很高效或者按照我們要求的去執(zhí)行任務(wù)。例如，經(jīng)過partition處理后，一個節(jié)點的reducer分配到了20條記錄，另一個卻分配道了10W萬條，試想，這種情況效率如何。又或者，我們想要處理后得到的文件按照一定的規(guī)律進(jìn)行輸出，假設(shè)有兩個reducer，我們想要最終結(jié)果中part-00000中存儲的是"h"開頭的記錄的結(jié)果,part-00001中存儲其他開頭的結(jié)果，這些默認(rèn)的partitioner是做不到的。所以需要我們自己定制partition來根據(jù)自己的要求，選擇記錄的reducer。自定義partitioner很簡單，只要自定義一個類，并且繼承Partitioner類，重寫其getPartition方法就好了，在使用的時候通過調(diào)用Job的setPartitionerClass指定一下即可

??????? Map的結(jié)果，會通過partition分發(fā)到Reducer上。Mapper的結(jié)果，可能送到Combiner做合并，Combiner在系統(tǒng)中并沒有自己的基類，而是用Reducer作為Combiner的基類，他們對外的功能是一樣的，只是使用的位置和使用時的上下文不太一樣而已。Mapper最終處理的鍵值對<key, value>，是需要送到Reducer去合并的，合并的時候，有相同key的鍵/值對會送到同一個Reducer那。哪個key到哪個Reducer的分配過程，是由Partitioner規(guī)定的。它只有一個方法，

??????? getPartition(Text key, Text value, int numPartitions)

輸入是Map的結(jié)果對<key, value>和Reducer的數(shù)目，輸出則是分配的Reducer（整數(shù)編號）。就是指定Mappr輸出的鍵值對到哪一個reducer上去。系統(tǒng)缺省的Partitioner是HashPartitioner，它以key的Hash值對Reducer的數(shù)目取模，得到對應(yīng)的Reducer。這樣保證如果有相同的key值，肯定被分配到同一個reducre上。如果有N個reducer，編號就為0,1,2,3……(N-1)。

（2）Partition的使用
??????? 分區(qū)出現(xiàn)的必要性，如何使用Hadoop產(chǎn)生一個全局排序的文件？最簡單的方法就是使用一個分區(qū)，但是該方法在處理大型文件時效率極低，因為一臺機器必須處理所有輸出文件，從而完全喪失了MapReduce所提供的并行架構(gòu)的優(yōu)勢。事實上我們可以這樣做，首先創(chuàng)建一系列排好序的文件；其次，串聯(lián)這些文件（類似于歸并排序）；最后得到一個全局有序的文件。主要的思路是使用一個partitioner來描述全局排序的輸出。比方說我們有1000個1-10000的數(shù)據(jù)，跑10個ruduce任務(wù)， 如果我們運行進(jìn)行partition的時候，能夠?qū)⒃?-1000中數(shù)據(jù)的分配到第一個reduce中，1001-2000的數(shù)據(jù)分配到第二個reduce中，以此類推。即第n個reduce所分配到的數(shù)據(jù)全部大于第n-1個reduce中的數(shù)據(jù)。這樣，每個reduce出來之后都是有序的了，我們只要cat所有的輸出文件，變成一個大的文件，就都是有序的了

基本思路就是這樣，但是現(xiàn)在有一個問題，就是數(shù)據(jù)的區(qū)間如何劃分，在數(shù)據(jù)量大，還有我們并不清楚數(shù)據(jù)分布的情況下。一個比較簡單的方法就是采樣，假如有一億的數(shù)據(jù)，我們可以對數(shù)據(jù)進(jìn)行采樣，如取10000個數(shù)據(jù)采樣，然后對采樣數(shù)據(jù)分區(qū)間。在Hadoop中，patition我們可以用TotalOrderPartitioner替換默認(rèn)的分區(qū)。然后將采樣的結(jié)果傳給他，就可以實現(xiàn)我們想要的分區(qū)。在采樣時，我們可以使用hadoop的幾種采樣工具，RandomSampler,InputSampler,IntervalSampler。

?????? 這樣，我們就可以對利用分布式文件系統(tǒng)進(jìn)行大數(shù)據(jù)量的排序了，我們也可以重寫Partitioner類中的compare函數(shù)，來定義比較的規(guī)則，從而可以實現(xiàn)字符串或其他非數(shù)字類型的排序，也可以實現(xiàn)二次排序乃至多次排序。

2、MapReduce中分組的概念和使用
??? 分區(qū)的目的是根據(jù)Key值決定Mapper的輸出記錄被送到哪一個Reducer上去處理。而分組的就比較好理解了。筆者認(rèn)為，分組就是與記錄的Key相關(guān)。在同一個分區(qū)里面，具有相同Key值的記錄是屬于同一個分組的。

3、MapReduce中Combiner的使用
??????? 很多MapReduce程序受限于集群上可用的帶寬，所以它會盡力最小化需要在map和reduce任務(wù)之間傳輸?shù)闹虚g數(shù)據(jù)。Hadoop允許用戶聲明一個combiner function來處理map的輸出，同時把自己對map的處理結(jié)果作為reduce的輸入。因為combiner function本身只是一種優(yōu)化，hadoop并不保證對于某個map輸出，這個方法會被調(diào)用多少次。換句話說，不管combiner function被調(diào)用多少次，對應(yīng)的reduce輸出結(jié)果都應(yīng)該是一樣的。

　　下面我們以《權(quán)威指南》的例子來加以說明，假設(shè)1950年的天氣數(shù)據(jù)讀取是由兩個map完成的，其中第一個map的輸出如下：
　　(1950, 0)
　　(1950, 20)
　　(1950, 10)

第二個map的輸出為：
?????? (1950, 25)
?????? (1950, 15)

而reduce得到的輸入為：(1950, [0, 20, 10, 25, 15])， 輸出為：(1950, 25)

　　由于25是集合中的最大值，我們可以使用一個類似于reduce function的combiner function來找出每個map輸出中的最大值，這樣的話，reduce的輸入就變成了：
　　(1950, [20, 25])

　　各個funciton 對溫度值的處理過程可以表示如下：max(0, 20, 10, 25, 15) =max(max(0, 20, 10), max(25, 15)) = max(20, 25) = 25

　　注意：并不是所有的函數(shù)都擁有這個屬性的（有這個屬性的函數(shù)我們稱之為commutative和associative），例如，如果我們要計算平均溫度，就不能這樣使用combiner function，因為mean(0, 20, 10, 25, 15) =14，而mean(mean(0, 20, 10),mean(25, 15)) = mean(10, 20) = 15

　　combiner function并不能取代reduce function（因為仍然需要reduce function處理來自不同map的帶有相同key的記錄）。但是他可以幫助減少需要在map和reduce之間傳輸?shù)臄?shù)據(jù)，就為這一點combiner function就值得考慮使用。

4、Shuffle階段排序流程詳解
??????? 我們首先看一下MapReduce中的排序的總體流程。

??????? MapReduce框架會確保每一個Reducer的輸入都是按Key進(jìn)行排序的。一般，將排序以及Map的輸出傳輸?shù)絉educe的過程稱為混洗（shuffle)。每一個Map都包含一個環(huán)形的緩存，默認(rèn)100M，Map首先將輸出寫到緩存當(dāng)中。當(dāng)緩存的內(nèi)容達(dá)到“閾值”時（閾值默認(rèn)的大小是緩存的80%），一個后臺線程負(fù)責(zé)將結(jié)果寫到硬盤，這個過程稱為“spill”。Spill過程中，Map仍可以向緩存寫入結(jié)果，如果緩存已經(jīng)寫滿，那么Map進(jìn)行等待。

Spill的具體過程如下：首先，后臺線程根據(jù)Reducer的個數(shù)將輸出結(jié)果進(jìn)行分組，每一個分組對應(yīng)一個Reducer。其次，對于每一個分組后臺線程對輸出結(jié)果的Key進(jìn)行排序。在排序過程中，如果有Combiner函數(shù)，則對排序結(jié)果進(jìn)行Combiner函數(shù)進(jìn)行調(diào)用。每一次spill都會在硬盤產(chǎn)生一個spill文件。因此，一個Map task有可能會產(chǎn)生多個spill文件，當(dāng)Map寫出最后一個輸出時，會將所有的spill文件進(jìn)行合并與排序，輸出最終的結(jié)果文件。在這個過程中Combiner函數(shù)仍然會被調(diào)用。從整個過程來看，Combiner函數(shù)的調(diào)用次數(shù)是不確定的。下面我們重點分析下Shuffle階段的排序過程：

??????? Shuffle階段的排序可以理解成兩部分，一個是對spill進(jìn)行分區(qū)時，由于一個分區(qū)包含多個key值，所以要對分區(qū)內(nèi)的<key,value>按照key進(jìn)行排序，即key值相同的一串<key,value>存放在一起，這樣一個partition內(nèi)按照key值整體有序了。

??????? 第二部分并不是排序，而是進(jìn)行merge，merge有兩次，一次是map端將多個spill 按照分區(qū)和分區(qū)內(nèi)的key進(jìn)行merge，形成一個大的文件。第二次merge是在reduce端，進(jìn)入同一個reduce的多個map的輸出 merge在一起，該merge理解起來有點復(fù)雜，最終不是形成一個大文件，而且期間數(shù)據(jù)在內(nèi)存和磁盤上都有。所以shuffle階段的merge并不是嚴(yán)格的排序意義，只是將多個整體有序的文件merge成一個大的文件，由于不同的task執(zhí)行map的輸出會有所不同，所以merge后的結(jié)果不是每次都相同，不過還是嚴(yán)格要求按照分區(qū)劃分，同時每個分區(qū)內(nèi)的具有相同key的<key,value>對挨在一起。

??????? Shuffle排序綜述：如果只定義了map函數(shù)，沒有定義reduce函數(shù)，那么輸入數(shù)據(jù)經(jīng)過shuffle的排序后，結(jié)果為key值相同的輸出挨在一起，且key值小的一定在前面，這樣整體來看key值有序（宏觀意義的，不一定是按從大到小，因為如果采用默認(rèn)的HashPartitioner，則key 的hash值相等的在一個分區(qū)，如果key為IntWritable的話，每個分區(qū)內(nèi)的key會排序好的），而每個key對應(yīng)的value不是有序的。

5、MapReduce中輔助排序的原理與實現(xiàn)
（1）任務(wù)
我們需要把內(nèi)容如下的sample.txt文件處理為下面文件：

源文件：Sample.txt

bbb 654

ccc 534

ddd 423

aaa 754

bbb 842

ccc 120

ddd 219

aaa 344

bbb 214

ccc 547

ddd 654

aaa 122

bbb 102

ccc 479

ddd 742

aaa 146

目標(biāo)：part-r-00000

aaa 122

bbb 102

ccc 120

ddd 219

（2）工作原理
?? 過程導(dǎo)引：
?? 1、定義包含記錄值和自然值的組合鍵，本例中為MyPariWritable.

?? 2、自定義鍵的比較器（comparator）來根據(jù)組合鍵對記錄進(jìn)行排序，即同時利用自然鍵和自然值進(jìn)行排序。（aaa 122組合為一個鍵）。

?? 3、針對組合鍵的Partitioner（本示例使用默認(rèn)的hashPartitioner）和分組comparator在進(jìn)行分區(qū)和分組時均只考慮自然鍵。

?? 詳細(xì)過程：
首先在map階段，使用job.setInputFormatClass定義的InputFormat將輸入的數(shù)據(jù)集分割成小數(shù)據(jù)塊splites，同時InputFormat提供一個RecordReder的實現(xiàn)。本例子中使用的是TextInputFormat，他提供的RecordReder會將文本的一行的行號作為key，這一行的文本作為value。這就是自定義Map的輸入是<LongWritable, Text>的原因。然后調(diào)用自定義Map的map方法，將一個個<LongWritable, Text>對輸入給Map的map方法。注意輸出應(yīng)該符合自定義Map中定義的輸出< MyPariWritable, NullWritable>。最終是生成一個List< MyPariWritable, NullWritable>。在map階段的最后，會先調(diào)用job.setPartitionerClass對這個List進(jìn)行分區(qū)，每個分區(qū)映射到一個reducer。每個分區(qū)內(nèi)又調(diào)用job.setSortComparatorClass設(shè)置的key比較函數(shù)類排序。可以看到，這本身就是一個二次排序。在reduce階段，reducer接收到所有映射到這個reducer的map輸出后，也是會調(diào)用job.setSortComparatorClass設(shè)置的key比較函數(shù)類對所有數(shù)據(jù)對排序。然后開始構(gòu)造一個key對應(yīng)的value迭代器。這時就要用到分組，使用jobjob.setGroupingComparatorClass設(shè)置的分組函數(shù)類。只要這個比較器比較的兩個key相同，他們就屬于同一個組（本例中由于要求得每一個分區(qū)內(nèi)的最小值，因此比較MyPariWritable類型的Key時，只需要比較自然鍵，這樣就能保證只要兩個MyPariWritable的自然鍵相同，則它們被送到Reduce端時候的Key就認(rèn)為在相同的分組，由于該分組的Key只取分組中的第一個，而這些數(shù)據(jù)已經(jīng)按照自定義MyPariWritable比較器排好序，則第一個Key正好包含了每一個自然鍵對應(yīng)的最小值），它們的value放在一個value迭代器，而這個迭代器的key使用屬于同一個組的所有key的第一個key。最后就是進(jìn)入Reducer的reduce方法，reduce方法的輸入是所有的key和它的value迭代器。同樣注意輸入與輸出的類型必須與自定義的Reducer中聲明的一致。

轉(zhuǎn)載于:https://my.oschina.net/u/2000675/blog/654720

總結(jié)

以上是生活随笔為你收集整理的Hadoop Mapreduce分区、分组、二次排序过程详解的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。