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

我在15年處理大數(shù)據(jù)的時(shí)候還都是使用MapReduce, 隨著時(shí)間的推移, 計(jì)算工具的發(fā)展, 內(nèi)存越來(lái)越便宜, 計(jì)算方式也有了極大的改變. 到現(xiàn)在再做大數(shù)據(jù)開(kāi)發(fā)的好多同學(xué)都是直接使用spark, hive等工具, 很少有再寫(xiě)MapReduce的了.
這里整理一下MapReduce中經(jīng)常用到的二次排序的方法, 全當(dāng)復(fù)習(xí).

簡(jiǎn)介

二次排序(secondary sort)問(wèn)題是指在Reduce階段對(duì)某個(gè)鍵關(guān)聯(lián)的值排序. 利用二次排序技術(shù),可以對(duì)傳入Reduce的值完成 升序/降序 排序.
MapReduce框架會(huì)自動(dòng)對(duì)Map生成的鍵完成排序. 所以, 在啟動(dòng)Reduce之前,中間文件 key-value 是按照key有序的(而不是按照值有序). 它們的值得順序有可能是任意的.

二次排序解決方案

對(duì)Reduce中的值排序至少有兩種方案, 這兩種方案在MapReduce/Hadoop 和 Spark框架中都可以使用.

• 第一種方案是讓Reduce讀取和緩存給定key的所有的value, 然后在Reduce中對(duì)這些值完成排序.(例如: 把一個(gè)key對(duì)應(yīng)的所有value放到一個(gè)Array或List中,再排序). 但是這種方式有局限性, 如果數(shù)據(jù)量較少還可以使用,如果數(shù)據(jù)量太大,一個(gè)Reduce中放不下所有的值,就會(huì)導(dǎo)致內(nèi)存溢出(OutOfMemory).
• 第二種方式是使用MapReduce框架來(lái)對(duì)值進(jìn)行排序. 因?yàn)镸apReduce框架會(huì)自動(dòng)對(duì)Map生成的文件的key進(jìn)行排序, 所以我們把需要排序的value增加到這個(gè)key上,這樣讓框架對(duì)這個(gè)new_key進(jìn)行排序,來(lái)實(shí)現(xiàn)我們的目標(biāo).

第二種方法小結(jié):

使用值鍵轉(zhuǎn)換設(shè)計(jì)模式:構(gòu)造一個(gè)組合的中間key,new_key(k, v1), 其中v1是次鍵(secondary key).

讓MapReduce執(zhí)行框架完成排序.

重寫(xiě)分區(qū)器,使組合鍵(k, v1) 按照之前單獨(dú)的 k 進(jìn)行分區(qū).

示例

假設(shè)有一組科學(xué)實(shí)驗(yàn)的溫度數(shù)據(jù)如下:
有4列分別為: 年, 月, 日, 溫度.

2000,12,04,10 2000,11,01,20 2000,12,02,-20 2000,11,07,30 2000,11,24,-40 2000,01,12,10 ...

我們需要輸出每一個(gè)年-月的溫度,并且值按照升序排序.
所以輸出如下:

(2000-11),[-40,20,30] (2000-01),[10] (2000-12),[-20,10]

MapReduce二次排序?qū)崿F(xiàn)細(xì)節(jié)

要實(shí)現(xiàn)二次排序的特性,還需要一些java的插件類(lèi), 去告訴MapReduce框架一些信息:

• 如何對(duì)Reduce的鍵排序.
• 如何對(duì)Map產(chǎn)出的數(shù)據(jù)進(jìn)行分區(qū),進(jìn)到不同的Reduce.
• 如何對(duì)Reduce中的數(shù)據(jù)進(jìn)行分組.

組合鍵的排序順序

要實(shí)現(xiàn)二次排序, 我們需要控制組合鍵的排序順序,以及Reduce處理鍵的順序.
首先組合鍵的組成由(年-月 + 溫度)一起組成, 如下圖:


把temperature的數(shù)據(jù)放到鍵中之后, 我們還要指定這個(gè)組合鍵排序方式. 使用DateTemperaturePair對(duì)象保存組合鍵, 重寫(xiě)其compareTo()方法指定排序順序.
Hadoop中,如果需要持久存儲(chǔ)定制數(shù)據(jù)類(lèi)型(如DateTemperaturePair),必須實(shí)現(xiàn)Writable接口. 如果要比較定制數(shù)據(jù)類(lèi)型, 他們還必須實(shí)現(xiàn)另外一個(gè)接口WritableComparable. 示例代碼如下:

import org.apache.hadoop.io.Writable; import org.apache.hadoop.io.WritableComparable; ... public class DateTemperaturePair implements Writable, WritableComparable<DateTemperaturePair> {private Text yearMonth = new Text(); //自然鍵private Text day = new Text();private IntWritable temperature = new IntWritable(); // 次鍵...@Override/*** 這個(gè)比較器將控制鍵的排序順序* /public int compareTo(DateTemperaturePair pair) {int compareValue = this.yearMonth.compareTo(pair.getYearMonth());if (compareValue == 0) {compareValue = temperature.compareTo(pair.getTemperature());}return compareValue; //升序排序//return -1 * compareValue; //降序排序} }

定制分區(qū)器

分區(qū)器默認(rèn)會(huì)根據(jù)Map產(chǎn)出的key來(lái)決定數(shù)據(jù)進(jìn)到哪個(gè)Reduce.
在這里,我們需要根據(jù)yearMonth來(lái)分區(qū)把數(shù)據(jù)入到不同的Reduce中, 但是我們的鍵已經(jīng)變成了(yearMonth + temperature)的組合了. 所以需要定制分區(qū)器來(lái)根據(jù)yearMonth進(jìn)行數(shù)據(jù)分區(qū),把相同的yearMonth入到一個(gè)Reduce中. 代碼如下:

import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner; public class DateTemperaturePartitioner extends Partitioner<DateTemperaturePair, Text> {@Overridepublic int getPartition(DatetemperaturePair pair, Text text, int numberOfPartitions) {//確保分區(qū)數(shù)非負(fù)return math.abs(pair.getYearMonth().hashCode() % numberOfPartitions);} }

Hadoop提供了一個(gè)插件體系,允許在框架中注入定制分區(qū)器代碼. 我們?cè)隍?qū)動(dòng)累中完成這個(gè)工作,如下:

import org.apache.hadoop.mapreduce.Job; ... Job job = ...; ... job.setPartitionerClass(TemperaturePartitioner.class);

分組比較器

分組比較器會(huì)控制哪些鍵要分組到一個(gè)Reduce.reduce()方法中調(diào)用.
默認(rèn)是按照key分配, 這里我們期望的是按照組合key(yearMonth + temperature) 中的yearMonth分配, 所以需要重寫(xiě)分組方法.
如下:

import org.apache.hadoop.io.WritableComparable; import org.apache.hadoop.io.WritableComparator; public class DateTemperatureGroupingComparator extends WritableComparator {public DateTemperatureGroupingComparator() {super(DateTemperaturePair.class, true);}@Override/*** 比較器控制哪些鍵要分組到一個(gè)reduce()方法調(diào)用*/public int compare(WritableComparable wc1, WritableComparable wc2) {DateTemperaturePair pair = (DateTemperaturePair) wc1;DateTemperaturePair pair2 = (DateTemperaturePair) wc12;return pair.getYearMonth().compareTo(pair2.getYearMonth());} }

在驅(qū)動(dòng)類(lèi)中注冊(cè)比較器:
job.setGroupingComparatorClass(YearMonthGroupingComparator.class);

使用插件的數(shù)據(jù)流



原理總結(jié)

MapReduce框架默認(rèn)會(huì)按照key來(lái)進(jìn)行分區(qū),排序,分組.
我們需要排序的時(shí)候使用key+value所以我們把key變成了新key, (firstkey, secondkey) 對(duì)應(yīng)為(yearMonth, 溫度) .

但是又不想在分區(qū) 和 分組的時(shí)候使用新key, 所以自己寫(xiě)了Partitioner 和 GroupingComparator 來(lái)指定使用組合key中的firstkey來(lái)分區(qū),分組.

--Posted from Rpc

轉(zhuǎn)載于:https://my.oschina.net/wangt10/blog/3050044

總結(jié)

以上是生活随笔為你收集整理的详细讲解MapReduce二次排序过程的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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