2019獨角獸企業(yè)重金招聘Python工程師標準>>>

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

簡介

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

二次排序解決方案

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

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

第二種方法小結:

使用值鍵轉換設計模式:構造一個組合的中間key,new_key(k, v1), 其中v1是次鍵(secondary key).

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

重寫分區(qū)器,使組合鍵(k, v1) 按照之前單獨的 k 進行分區(qū).

示例

假設有一組科學實驗的溫度數(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 ...

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

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

MapReduce二次排序實現(xiàn)細節(jié)

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

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

組合鍵的排序順序

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


把temperature的數(shù)據(jù)放到鍵中之后, 我們還要指定這個組合鍵排序方式. 使用DateTemperaturePair對象保存組合鍵, 重寫其compareTo()方法指定排序順序.
Hadoop中,如果需要持久存儲定制數(shù)據(jù)類型(如DateTemperaturePair),必須實現(xiàn)Writable接口. 如果要比較定制數(shù)據(jù)類型, 他們還必須實現(xiàn)另外一個接口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/*** 這個比較器將控制鍵的排序順序* /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ū)器默認會根據(jù)Map產出的key來決定數(shù)據(jù)進到哪個Reduce.
在這里,我們需要根據(jù)yearMonth來分區(qū)把數(shù)據(jù)入到不同的Reduce中, 但是我們的鍵已經變成了(yearMonth + temperature)的組合了. 所以需要定制分區(qū)器來根據(jù)yearMonth進行數(shù)據(jù)分區(qū),把相同的yearMonth入到一個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ù)非負return math.abs(pair.getYearMonth().hashCode() % numberOfPartitions);} }

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

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

分組比較器

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

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

在驅動類中注冊比較器:
job.setGroupingComparatorClass(YearMonthGroupingComparator.class);

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



原理總結

MapReduce框架默認會按照key來進行分區(qū),排序,分組.
我們需要排序的時候使用key+value所以我們把key變成了新key, (firstkey, secondkey) 對應為(yearMonth, 溫度) .

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

--Posted from Rpc

轉載于:https://my.oschina.net/wangt10/blog/3050044

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。