hadoop二次排序的原理和實(shí)現(xiàn)方法
默認(rèn)情況下,Map輸出的結(jié)果會(huì)對(duì)Key進(jìn)行默認(rèn)的排序,但是有時(shí)候需要對(duì)Key排序的同時(shí)還需要對(duì)Value進(jìn)行排序,這時(shí)候就要用到二次排序了。下面我們來(lái)說(shuō)說(shuō)二次排序
1、二次排序原理
我們把二次排序分為以下幾個(gè)階段
Map起始階段
在Map階段,使用job.setInputFormatClass()定義的InputFormat,將輸入的數(shù)據(jù)集分割成小數(shù)據(jù)塊split,同時(shí)InputFormat提供一個(gè)RecordReader的實(shí)現(xiàn)。在這里我們使用的是TextInputFormat,它提供的RecordReader會(huì)將文本的行號(hào)作為Key,這一行的文本作為Value。這就是自定 Mapper的輸入是<LongWritable,Text> 的原因。然后調(diào)用自定義Mapper的map方法,將一個(gè)個(gè)<LongWritable,Text>鍵值對(duì)輸入給Mapper的map方法
Map最后階段
在Map階段的最后,會(huì)先調(diào)用job.setPartitionerClass()對(duì)這個(gè)Mapper的輸出結(jié)果進(jìn)行分區(qū),每個(gè)分區(qū)映射到一個(gè)Reducer。每個(gè)分區(qū)內(nèi)又調(diào)用job.setSortComparatorClass()設(shè)置的Key比較函數(shù)類排序。可以看到,這本身就是一個(gè)二次排序。如果沒有通過job.setSortComparatorClass()設(shè)置 Key比較函數(shù)類,則使用Key實(shí)現(xiàn)的compareTo()方法
Reduce階段
在Reduce階段,reduce()方法接受所有映射到這個(gè)Reduce的map輸出后,也會(huì)調(diào)用job.setSortComparatorClass()方法設(shè)置的Key比較函數(shù)類,對(duì)所有數(shù)據(jù)進(jìn)行排序。然后開始構(gòu)造一個(gè)Key對(duì)應(yīng)的Value迭代器。這時(shí)就要用到分組,使用 job.setGroupingComparatorClass()方法設(shè)置分組函數(shù)類。只要這個(gè)比較器比較的兩個(gè)Key相同,它們就屬于同一組,它們的 Value放在一個(gè)Value迭代器,而這個(gè)迭代器的Key使用屬于同一個(gè)組的所有Key的第一個(gè)Key。最后就是進(jìn)入Reducer的 reduce()方法,reduce()方法的輸入是所有的Key和它的Value迭代器,同樣注意輸入與輸出的類型必須與自定義的Reducer中聲明的一致
接下來(lái)我們通過示例,可以很直觀的了解二次排序的原理
輸入文件 sort.txt 內(nèi)容為
40 20 40 10 40 30 40 5 30 30 30 20 30 10 30 40 50 20 50 50 50 10 50 60
輸出文件的內(nèi)容(從小到大排序)如下
30 10 30 20 30 30 30 40 -------- 40 5 40 10 40 20 40 30 -------- 50 10 50 20 50 50 50 60
從輸出的結(jié)果可以看出Key實(shí)現(xiàn)了從小到大的排序,同時(shí)相同Key的Value也實(shí)現(xiàn)了從小到大的排序,這就是二次排序的結(jié)果
2、二次排序的具體流程
在本例中要比較兩次。先按照第一字段排序,然后再對(duì)第一字段相同的按照第二字段排序。根據(jù)這一點(diǎn),我們可以構(gòu)造一個(gè)復(fù)合類IntPair ,它有兩個(gè)字段,先利用分區(qū)對(duì)第一字段排序,再利用分區(qū)內(nèi)的比較對(duì)第二字段排序。二次排序的流程分為以下幾步。
在本例中要比較兩次。先按照第一字段排序,然后再對(duì)第一字段相同的按照第二字段排序。根據(jù)這一點(diǎn),我們可以構(gòu)造一個(gè)復(fù)合類IntPair ,它有兩個(gè)字段,先利用分區(qū)對(duì)第一字段排序,再利用分區(qū)內(nèi)的比較對(duì)第二字段排序。二次排序的流程分為以下幾步。
1、自定義 key
所有自定義的key應(yīng)該實(shí)現(xiàn)接口WritableComparable,因?yàn)樗强尚蛄谢牟⑶铱杀容^的。WritableComparable 的內(nèi)部方法如下所示
// 反序列化,從流中的二進(jìn)制轉(zhuǎn)換成IntPair public void readFields(DataInput in) throws IOException // 序列化,將IntPair轉(zhuǎn)化成使用流傳送的二進(jìn)制 public void write(DataOutput out) // key的比較 public int compareTo(IntPair o) // 默認(rèn)的分區(qū)類 HashPartitioner,使用此方法 public int hashCode() // 默認(rèn)實(shí)現(xiàn) public boolean equals(Object right)
2、自定義分區(qū)
自定義分區(qū)函數(shù)類FirstPartitioner,是key的第一次比較,完成對(duì)所有key的排序。
public static class FirstPartitioner extends Partitioner< IntPair,IntWritable>
在job中使用setPartitionerClasss()方法設(shè)置Partitioner
job.setPartitionerClasss(FirstPartitioner.Class);
3、Key的比較類
這是Key的第二次比較,對(duì)所有的Key進(jìn)行排序,即同時(shí)完成IntPair中的first和second排序。該類是一個(gè)比較器,可以通過兩種方式實(shí)現(xiàn)。
1) 繼承WritableComparator。
public static class KeyComparator extends WritableComparator
必須有一個(gè)構(gòu)造函數(shù),并且重載以下方法。
public int compare(WritableComparable w1, WritableComparable w2)
2) 實(shí)現(xiàn)接口 RawComparator。
上面兩種實(shí)現(xiàn)方式,在Job中,可以通過setSortComparatorClass()方法來(lái)設(shè)置Key的比較類。
job.setSortComparatorClass(KeyComparator.Class);
注意:如果沒有使用自定義的SortComparator類,則默認(rèn)使用Key中compareTo()方法對(duì)Key排序。
4、定義分組類函數(shù)
在Reduce階段,構(gòu)造一個(gè)與 Key 相對(duì)應(yīng)的 Value 迭代器的時(shí)候,只要first相同就屬于同一個(gè)組,放在一個(gè)Value迭代器。定義這個(gè)比較器,可以有兩種方式。
1) 繼承 WritableComparator。
public static class GroupingComparator extends WritableComparator
必須有一個(gè)構(gòu)造函數(shù),并且重載以下方法。
public int compare(WritableComparable w1, WritableComparable w2)
2) 實(shí)現(xiàn)接口 RawComparator。
上面兩種實(shí)現(xiàn)方式,在 Job 中,可以通過 setGroupingComparatorClass()方法來(lái)設(shè)置分組類。
job.setGroupingComparatorClass(GroupingComparator.Class);
另外注意的是,如果reduce的輸入與輸出不是同一種類型,則 Combiner和Reducer 不能共用 Reducer 類,因?yàn)?/p>
Combiner 的輸出是 reduce 的輸入。除非重新定義一個(gè)Combiner。
3、代碼實(shí)現(xiàn)
Hadoop的example包中自帶了一個(gè)MapReduce的二次排序算法,下面對(duì) example包中的二次排序進(jìn)行改進(jìn)
package com.buaa; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; import org.apache.hadoop.io.WritableComparable; /** * @ProjectName SecondarySort * @PackageName com.buaa * @ClassName IntPair * @Description 將示例數(shù)據(jù)中的key/value封裝成一個(gè)整體作為Key,同時(shí)實(shí)現(xiàn) WritableComparable接口并重寫其方法 * @Author 劉吉超 * @Date 2016-06-07 22:31:53 */ public class IntPair implements WritableComparable<IntPair>{ private int first; private int second; public IntPair(){ } public IntPair(int left, int right){ set(left, right); } public void set(int left, int right){ first = left; second = right; } @Override public void readFields(DataInput in) throws IOException{ first = in.readInt(); second = in.readInt(); } @Override public void write(DataOutput out) throws IOException{ out.writeInt(first); out.writeInt(second); } @Override public int compareTo(IntPair o) { if (first != o.first){ return first < o.first ? -1 : 1; }else if (second != o.second){ return second < o.second ? -1 : 1; }else{ return 0; } } @Override public int hashCode(){ return first * 157 + second; } @Override public boolean equals(Object right){ if (right == null) return false; if (this == right) return true; if (right instanceof IntPair){ IntPair r = (IntPair) right; return r.first == first && r.second == second; }else{ return false; } } public int getFirst(){ return first; } public int getSecond(){ return second; } } package com.buaa; import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.WritableComparable; import org.apache.hadoop.io.WritableComparator; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Partitioner; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; /** * @ProjectName SecondarySort * @PackageName com.buaa * @ClassName SecondarySort * @Description TODO * @Author 劉吉超 * @Date 2016-06-07 22:40:37 */ @SuppressWarnings("deprecation") public class SecondarySort { public static class Map extends Mapper<LongWritable, Text, IntPair, IntWritable> { public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); StringTokenizer tokenizer = new StringTokenizer(line); int left = 0; int right = 0; if (tokenizer.hasMoreTokens()) { left = Integer.parseInt(tokenizer.nextToken()); if (tokenizer.hasMoreTokens()) right = Integer.parseInt(tokenizer.nextToken()); context.write(new IntPair(left, right), new IntWritable(right)); } } } /* * 自定義分區(qū)函數(shù)類FirstPartitioner,根據(jù) IntPair中的first實(shí)現(xiàn)分區(qū) */ public static class FirstPartitioner extends Partitioner<IntPair, IntWritable>{ @Override public int getPartition(IntPair key, IntWritable value,int numPartitions){ return Math.abs(key.getFirst() * 127) % numPartitions; } } /* * 自定義GroupingComparator類,實(shí)現(xiàn)分區(qū)內(nèi)的數(shù)據(jù)分組 */ @SuppressWarnings("rawtypes") public static class GroupingComparator extends WritableComparator{ protected GroupingComparator(){ super(IntPair.class, true); } @Override public int compare(WritableComparable w1, WritableComparable w2){ IntPair ip1 = (IntPair) w1; IntPair ip2 = (IntPair) w2; int l = ip1.getFirst(); int r = ip2.getFirst(); return l == r ? 0 : (l < r ? -1 : 1); } } public static class Reduce extends Reducer<IntPair, IntWritable, Text, IntWritable> { public void reduce(IntPair key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { for (IntWritable val : values) { context.write(new Text(Integer.toString(key.getFirst())), val); } } } public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException { // 讀取配置文件 Configuration conf = new Configuration(); // 判斷路徑是否存在,如果存在,則刪除 Path mypath = new Path(args[1]); FileSystem hdfs = mypath.getFileSystem(conf); if (hdfs.isDirectory(mypath)) { hdfs.delete(mypath, true); } Job job = new Job(conf, "secondarysort"); // 設(shè)置主類 job.setJarByClass(SecondarySort.class); // 輸入路徑 FileInputFormat.setInputPaths(job, new Path(args[0])); // 輸出路徑 FileOutputFormat.setOutputPath(job, new Path(args[1])); // Mapper job.setMapperClass(Map.class); // Reducer job.setReducerClass(Reduce.class); // 分區(qū)函數(shù) job.setPartitionerClass(FirstPartitioner.class); // 本示例并沒有自定義SortComparator,而是使用IntPair中compareTo方法進(jìn)行排序 job.setSortComparatorClass(); // 分組函數(shù) job.setGroupingComparatorClass(GroupingComparator.class); // map輸出key類型 job.setMapOutputKeyClass(IntPair.class); // map輸出value類型 job.setMapOutputValueClass(IntWritable.class); // reduce輸出key類型 job.setOutputKeyClass(Text.class); // reduce輸出value類型 job.setOutputValueClass(IntWritable.class); // 輸入格式 job.setInputFormatClass(TextInputFormat.class); // 輸出格式 job.setOutputFormatClass(TextOutputFormat.class); System.exit(job.waitForCompletion(true) ? 0 : 1); } }
總結(jié)
以上所述是小編給大家介紹的hadoop二次排序的原理和實(shí)現(xiàn)方法,希望對(duì)大家有所幫助,如果大家有任何疑問歡迎給我留言,小編會(huì)及時(shí)回復(fù)大家的!
相關(guān)文章
解決MobaXterm連接報(bào)錯(cuò)Network error:Connection timed 
這篇文章主要介紹了解決MobaXterm連接報(bào)錯(cuò)Network error:Connection timed out問題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-05-05vscode通過Remote SSH遠(yuǎn)程連接及離線配置的方法
這篇文章主要介紹了vscode通過Remote SSH遠(yuǎn)程連接及離線配置的方法,本文給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2021-03-03學(xué)編程選什么語(yǔ)言好?是PHP、Python還是Ruby?
這篇文章主要介紹了學(xué)編程選什么語(yǔ)言好?是PHP、Python還是Ruby?需要的朋友可以參考下2014-06-06科學(xué)知識(shí):二進(jìn)制、八進(jìn)制、十進(jìn)制、十六進(jìn)制轉(zhuǎn)換
這篇文章主要介紹了科學(xué)知識(shí):二進(jìn)制、八進(jìn)制、十進(jìn)制、十六進(jìn)制轉(zhuǎn)換,本文只介紹一些理論知識(shí),需要的朋友可以參考下2015-05-05win10環(huán)境安裝kettle與linux環(huán)境安裝kettle的詳細(xì)過程
kettle是一款免費(fèi)開源的、可視化的、國(guó)際上比較流行的、功能強(qiáng)大的ETL必備工具,在ETL這一方面做的還不錯(cuò),下面介紹一下基于win10操作系統(tǒng)安裝kettle和linux操作系統(tǒng)安裝kettle的詳細(xì)過程,感興趣的朋友跟隨小編一起看看吧2022-11-11