Hadoop日记Day18---MapReduce排序分组
本节所用到的数据下载地址为：http://pan.baidu.com/s/1bnfELmZjava
MapReduce的排序分组任务与要求
咱们知道排序分组是MapReduce中Mapper端的第四步，其中分组排序都是基于Key的，咱们能够经过下面这几个例子来体现出来。其中的数据和任务以下图1.1，1.2所示。算法
#首先按照第一列升序排列，当第一列相同时，第二列升序排列 3 3 3 2 3 1 2 2 2 1 1 1 ------------------- #结果 1 1 2 1 2 2 3 1 3 2 3 3
图 1.1 排序sql
#当第一列相同时，求出第二列的最小值 3 3 3 2 3 1 2 2 2 1 1 1 ------------------- #结果 3 1 2 1 1 1
图 1.2 分组数据库
1、 排序算法
1.1 MapReduce默认排序算法
使用MapReduce默认排序算法代码以下1.1所示，在代码中我将第一列做为键，第二列做为值。apache
1 package sort; 2 3 import java.io.IOException; 4 import java.net.URI; 5 6 import org.apache.hadoop.conf.Configuration; 7 import org.apache.hadoop.fs.FileStatus; 8 import org.apache.hadoop.fs.FileSystem; 9 import org.apache.hadoop.fs.Path; 10 import org.apache.hadoop.io.LongWritable; 11 import org.apache.hadoop.io.Text; 12 import org.apache.hadoop.mapreduce.Job; 13 import org.apache.hadoop.mapreduce.Mapper; 14 import org.apache.hadoop.mapreduce.Reducer; 15 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 16 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; 17 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 18 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; 19 import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner; 20 21 public class SortApp { 22 private static final String INPUT_PATH = "hdfs://hadoop:9000/newinput"; 23 private static final String OUT_PATH = "hdfs://hadoop:9000/newoutput"; 24 public static void main(String[] args) throws Exception { 25 Configuration conf=new Configuration(); 26 final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf); 27 final Path outpath = new Path(OUT_PATH); 28 if(fileSystem.exists(outpath)){ 29 fileSystem.delete(outpath,true); 30 } 31 32 final Job job = new Job(conf,SortApp.class.getSimpleName()); 33 34 //1.1 指定输入文件路径 35 FileInputFormat.setInputPaths(job, INPUT_PATH); 36 job.setInputFormatClass(TextInputFormat.class);//指定哪一个类用来格式化输入文件 37 38 //1.2指定自定义的Mapper类 39 job.setMapperClass(MyMapper.class); 40 job.setMapOutputKeyClass(LongWritable.class);//指定输出<k2,v2>的类型 41 job.setMapOutputValueClass(LongWritable.class); 42 43 //1.3 指定分区类 44 job.setPartitionerClass(HashPartitioner.class); 45 job.setNumReduceTasks(1); 46 47 //1.4 TODO 排序、分区 48 49 //1.5 TODO （可选）合并 50 51 //2.2 指定自定义的reduce类 52 job.setReducerClass(MyReducer.class); 53 job.setOutputKeyClass(LongWritable.class);//指定输出<k3,v3>的类型 54 job.setOutputValueClass(LongWritable.class); 55 56 //2.3 指定输出到哪里 57 FileOutputFormat.setOutputPath(job, outpath); 58 job.setOutputFormatClass(TextOutputFormat.class);//设定输出文件的格式化类 59 job.waitForCompletion(true);//把代码提交给JobTracker执行 60 } 61 static class MyMapper extends Mapper<LongWritable, Text,LongWritable,LongWritable>{ 62 63 @Override 64 protected void map( 65 LongWritable key, 66 Text value, 67 Mapper<LongWritable, Text, LongWritable, LongWritable>.Context context) 68 throws IOException, InterruptedException { 69 final String[] splited = value.toString().split("\t"); 70 final long k2 = Long.parseLong(splited[0]); 71 final long v2 = Long.parseLong(splited[1]); 72 context.write(new LongWritable(k2),new LongWritable(v2)); 73 } 74 } 75 static class MyReducer extends Reducer<LongWritable,LongWritable,LongWritable,LongWritable>{ 76 77 @Override 78 protected void reduce( 79 LongWritable k2, 80 Iterable<LongWritable> v2s, 81 Reducer<LongWritable, LongWritable, LongWritable, LongWritable>.Context context) 82 throws IOException, InterruptedException { 83 for(LongWritable v2:v2s){ 84 context.write(k2, v2); 85 } 86 } 87 } 88 }
代码 1.1数组
运行结果以下图1.3所示缓存
1 1 2 2 2 1 3 3 3 2 3 1
图 1.3网络
从上面图中运行结果能够看出，MapReduce默认排序算法只对Key进行了排序，并无对value进行排序，没有达到咱们的要求，因此要实现咱们的要求，还要咱们自定义一个排序算法app
1.2 自定义排序算法
从上面图中运行结果能够知道，MapReduce默认排序算法只对Key进行了排序，并无对value进行排序，没有达到咱们的要求，因此要实现咱们的要求，还要咱们自定义一个排序算法。在map和reduce阶段进行排序时，比较的是k2。v2是不参与排序比较的。若是要想让v2也进行排序，须要把k2和v2组装成新的类做为k 2 ，才能参与比较。因此在这里咱们新建一个新的类型NewK2类型来封装原来的k2和v2。代码如1.2所示。分布式
1 package sort; 2 3 import java.io.DataInput; 4 import java.io.DataOutput; 5 import java.io.IOException; 6 import java.net.URI; 7 8 import org.apache.hadoop.conf.Configuration; 9 import org.apache.hadoop.fs.FileSystem; 10 import org.apache.hadoop.fs.Path; 11 import org.apache.hadoop.io.LongWritable; 12 import org.apache.hadoop.io.Text; 13 import org.apache.hadoop.io.WritableComparable; 14 import org.apache.hadoop.mapreduce.Job; 15 import org.apache.hadoop.mapreduce.Mapper; 16 import org.apache.hadoop.mapreduce.Reducer; 17 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 18 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; 19 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 20 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; 21 import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner; 22 23 public class SortApp { 24 static final String INPUT_PATH = "hdfs://hadoop:9000/newinput"; 25 static final String OUT_PATH = "hdfs://hadoop:9000/newoutput"; 26 public static void main(String[] args) throws Exception{ 27 final Configuration configuration = new Configuration(); 28 final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), configuration); 29 if(fileSystem.exists(new Path(OUT_PATH))){ 30 fileSystem.delete(new Path(OUT_PATH), true); 31 } 32 final Job job = new Job(configuration, SortApp.class.getSimpleName()); 33 //1.1 指定输入文件路径 34 FileInputFormat.setInputPaths(job, INPUT_PATH); 35 job.setInputFormatClass(TextInputFormat.class);//指定哪一个类用来格式化输入文件 36 37 //1.2指定自定义的Mapper类 38 job.setMapperClass(MyMapper.class); 39 job.setMapOutputKeyClass(NewK2.class);//指定输出<k2,v2>的类型 40 job.setMapOutputValueClass(LongWritable.class); 41 42 //1.3 指定分区类 43 job.setPartitionerClass(HashPartitioner.class); 44 job.setNumReduceTasks(1); 45 46 //2.2 指定自定义的reduce类 47 job.setReducerClass(MyReducer.class); 48 job.setOutputKeyClass(LongWritable.class);//指定输出<k3,v3>的类型 49 job.setOutputValueClass(LongWritable.class); 50 51 //2.3 指定输出到哪里 52 FileOutputFormat.setOutputPath(job, new Path(OUT_PATH)); 53 job.setOutputFormatClass(TextOutputFormat.class);//设定输出文件的格式化类 54 job.waitForCompletion(true);//把代码提交给JobTracker执行 55 } 56 57 58 static class MyMapper extends Mapper<LongWritable, Text, NewK2, LongWritable>{ 59 protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,NewK2,LongWritable>.Context context) throws java.io.IOException ,InterruptedException { 60 final String[] splited = value.toString().split("\t"); 61 final NewK2 k2 = new NewK2(Long.parseLong(splited[0]), Long.parseLong(splited[1])); 62 final LongWritable v2 = new LongWritable(Long.parseLong(splited[1])); 63 context.write(k2, v2); 64 }; 65 } 66 67 static class MyReducer extends Reducer<NewK2, LongWritable, LongWritable, LongWritable>{ 68 protected void reduce(NewK2 k2, java.lang.Iterable<LongWritable> v2s, org.apache.hadoop.mapreduce.Reducer<NewK2,LongWritable,LongWritable,LongWritable>.Context context) throws java.io.IOException ,InterruptedException { 69 context.write(new LongWritable(k2.first), new LongWritable(k2.second)); 70 }; 71 } 72 73 /** 74 * 问：为何实现该类？ 75 * 答：由于原来的v2不能参与排序，把原来的k2和v2封装到一个类中，做为新的k2 76 * 77 */ 78 static class NewK2 implements WritableComparable<NewK2>{ 79 Long first; 80 Long second; 81 82 public NewK2(){} 83 84 public NewK2(long first, long second){ 85 this.first = first; 86 this.second = second; 87 } 88 89 90 @Override 91 public void readFields(DataInput in) throws IOException { 92 this.first = in.readLong(); 93 this.second = in.readLong(); 94 } 95 96 @Override 97 public void write(DataOutput out) throws IOException { 98 out.writeLong(first); 99 out.writeLong(second); 100 } 101 102 /** 103 * 当k2进行排序时，会调用该方法. 104 * 当第一列不一样时，升序；当第一列相同时，第二列升序 105 */ 106 @Override 107 public int compareTo(NewK2 o) { 108 final long minus = this.first - o.first; 109 if(minus !=0){ 110 return (int)minus; 111 } 112 return (int)(this.second - o.second); 113 } 114 115 @Override 116 public int hashCode() { 117 return this.first.hashCode()+this.second.hashCode(); 118 } 119 120 @Override 121 public boolean equals(Object obj) { 122 if(!(obj instanceof NewK2)){ 123 return false; 124 } 125 NewK2 oK2 = (NewK2)obj; 126 return (this.first==oK2.first)&&(this.second==oK2.second); 127 } 128 } 129 130 }
代码 1.2
从上面的代码中咱们能够发现，咱们的新类型NewK2实现了WritableComparable接口，其中该接口中有一个compareTo()方法，当对关键字进行比较会调用该方法，而咱们就在该方法中实现了咱们想要作的事。
运行结果以下图1.4所示。
1 1 2 1 2 2 3 1 3 2 3 3
图 1.4
2、分组算法
2.1 MapReduce默认分组
分组是在MapReduce中Mapper端的第四步，分组也是基于Key进行的，将相同key的value放到一个集合中去。还以上面排序代码为例，业务逻辑以下图2.1所示。在代码中以NewK2为关键字，每一个键都不相同，因此会将数据分为六组，这样就不能实现咱们的业务要求，但利用自定义类型NewK2，能够自定义排序算法的同时咱们也能够自定义分组算法。
#当第一列相同时，求出第二列的最小值
3 3
3 2
3 1
2 2
2 1
1 1
------------------- #结果 3 1 2 1 1 1
图 2.1
2.2 自定义分组比较器
因为业务要求分组是按照第一列分组，可是NewK2的比较规则决定了不能按照第一列分，只能自定义分组比较器，代码以下2.1所示。
1 package group; 2 3 import java.io.DataInput; 4 import java.io.DataOutput; 5 import java.io.IOException; 6 import java.net.URI; 7 8 import org.apache.hadoop.conf.Configuration; 9 import org.apache.hadoop.fs.FileSystem; 10 import org.apache.hadoop.fs.Path; 11 import org.apache.hadoop.io.LongWritable; 12 import org.apache.hadoop.io.RawComparator; 13 import org.apache.hadoop.io.Text; 14 import org.apache.hadoop.io.WritableComparable; 15 import org.apache.hadoop.io.WritableComparator; 16 import org.apache.hadoop.mapreduce.Job; 17 import org.apache.hadoop.mapreduce.Mapper; 18 import org.apache.hadoop.mapreduce.Reducer; 19 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 20 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; 21 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 22 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; 23 import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner; 24 25 public class GroupApp { 26 static final String INPUT_PATH = "hdfs://hadoop:9000/newinput"; 27 static final String OUT_PATH = "hdfs://hadoop:9000/newoutput"; 28 public static void main(String[] args) throws Exception{ 29 final Configuration configuration = new Configuration(); 30 31 final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), configuration); 32 if(fileSystem.exists(new Path(OUT_PATH))){ 33 fileSystem.delete(new Path(OUT_PATH), true); 34 } 35 final Job job = new Job(configuration, GroupApp.class.getSimpleName()); 36 37 //1.1 指定输入文件路径 38 FileInputFormat.setInputPaths(job, INPUT_PATH); 39 job.setInputFormatClass(TextInputFormat.class);//指定哪一个类用来格式化输入文件 40 41 //1.2指定自定义的Mapper类 42 job.setMapperClass(MyMapper.class); 43 job.setMapOutputKeyClass(NewK2.class);//指定输出<k2,v2>的类型 44 job.setMapOutputValueClass(LongWritable.class); 45 46 //1.3 指定分区类 47 job.setPartitionerClass(HashPartitioner.class); 48 job.setNumReduceTasks(1); 49 50 //1.4 TODO 排序、分区 51 job.setGroupingComparatorClass(MyGroupingComparator.class); 52 //1.5 TODO （可选）合并 53 54 //2.2 指定自定义的reduce类 55 job.setReducerClass(MyReducer.class); 56 job.setOutputKeyClass(LongWritable.class);//指定输出<k3,v3>的类型 57 job.setOutputValueClass(LongWritable.class); 58 59 //2.3 指定输出到哪里 60 FileOutputFormat.setOutputPath(job, new Path(OUT_PATH)); 61 job.setOutputFormatClass(TextOutputFormat.class);//设定输出文件的格式化类 62 job.waitForCompletion(true);//把代码提交给JobTracker执行 63 } 64 65 66 static class MyMapper extends Mapper<LongWritable, Text, NewK2, LongWritable>{ 67 protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,NewK2,LongWritable>.Context context) throws java.io.IOException ,InterruptedException { 68 final String[] splited = value.toString().split("\t"); 69 final NewK2 k2 = new NewK2(Long.parseLong(splited[0]), Long.parseLong(splited[1])); 70 final LongWritable v2 = new LongWritable(Long.parseLong(splited[1])); 71 context.write(k2, v2); 72 }; 73 } 74 75 static class MyReducer extends Reducer<NewK2, LongWritable, LongWritable, LongWritable>{ 76 protected void reduce(NewK2 k2, java.lang.Iterable<LongWritable> v2s, org.apache.hadoop.mapreduce.Reducer<NewK2,LongWritable,LongWritable,LongWritable>.Context context) throws java.io.IOException ,InterruptedException { 77 long min = Long.MAX_VALUE; 78 for (LongWritable v2 : v2s) { 79 if(v2.get()<min){ 80 min = v2.get(); 81 } 82 } 83 84 context.write(new LongWritable(k2.first), new LongWritable(min)); 85 }; 86 } 87 88 /** 89 * 问：为何实现该类？ 90 * 答：由于原来的v2不能参与排序，把原来的k2和v2封装到一个类中，做为新的k2 91 * 92 */ 93 static class NewK2 implements WritableComparable<NewK2>{ 94 Long first; 95 Long second; 96 97 public NewK2(){} 98 99 public NewK2(long first, long second){ 100 this.first = first; 101 this.second = second; 102 } 103 104 105 @Override 106 public void readFields(DataInput in) throws IOException { 107 this.first = in.readLong(); 108 this.second = in.readLong(); 109 } 110 111 @Override 112 public void write(DataOutput out) throws IOException { 113 out.writeLong(first); 114 out.writeLong(second); 115 } 116 117 /** 118 * 当k2进行排序时，会调用该方法. 119 * 当第一列不一样时，升序；当第一列相同时，第二列升序 120 */ 121 @Override 122 public int compareTo(NewK2 o) { 123 final long minus = this.first - o.first; 124 if(minus !=0){ 125 return (int)minus; 126 } 127 return (int)(this.second - o.second); 128 } 129 130 @Override 131 public int hashCode() { 132 return this.first.hashCode()+this.second.hashCode(); 133 } 134 135 @Override 136 public boolean equals(Object obj) { 137 if(!(obj instanceof NewK2)){ 138 return false; 139 } 140 NewK2 oK2 = (NewK2)obj; 141 return (this.first==oK2.first)&&(this.second==oK2.second); 142 } 143 } 144 145 static class MyGroupingComparator implements RawComparator<NewK2>{ 146 147 @Override 148 public int compare(NewK2 o1, NewK2 o2) { 149 return (int)(o1.first - o2.first); 150 } 151 152 @Override 153 public int compare(byte[] arg0, int arg1, int arg2, byte[] arg3, 154 int arg4, int arg5) { 155 return WritableComparator.compareBytes(arg0, arg1, 8, arg3, arg4, 8); 156 } 157 158 } 159 }
代码2.1
从上面的代码中咱们能够知道，咱们自定义了一个分组比较器MyGroupingComparator，该类实现了RawComparator接口，RawComparator又继承了Comparator接口，这两个接口的代码以下：
public interface RawComparator<T> extends Comparator<T> { public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2); }
public interface Comparator<T> { int compare(T o1, T o2); boolean equals(Object obj); }
在类MyGroupingComparator中分别对着两个接口中的方法进行了实现，RawComparator中的compare()方法是基于字节的比较，Comparator中的compare()方法是基于对象的比较。在该方法一共有六个参数，以下：
* @param arg0 表示第一个参与比较的字节数组
* @param arg1 表示第一个参与比较的字节数组的起始位置
* @param arg2 表示第一个参与比较的字节数组的偏移量
*
* @param arg3 表示第二个参与比较的字节数组
* @param arg4 表示第二个参与比较的字节数组的起始位置
* @param arg5 表示第二个参与比较的字节数组的偏移量
在于NewK2中存储着两个long类型，每一个long类型为8字节，.compareBytes()方法的参数以下：.compareBytes(arg0, arg1, 8, arg3, arg4, 8);由于比较的是第一列，因此读取的偏移量为8字节。因为咱们要求出每一分组的最小值，因此还重写Reduce方法，求出每一分租的最小值。最后的运行结果以下图2.1所示
1 1 2 1 3 1
图 2.1
3、MapReduce的一些算法
3.1 MapReduce中Shuffle过程
Shuffle是MapReduce过程的核心，了解Shuffle很是有助于理解MapReduce的工做原理。huffle的正常意思是洗牌或弄乱，可能你们更熟悉的是Java API里的Collections.shuffle(List)方法，它会随机地打乱参数list里的元素顺序。若是你不知道MapReduce里Shuffle是什么，那么请看这张图：
在该图中分为Map任务和Reduce任务两个部分，从map端到reduce端的红色和绿色的线表示数据流的一个过程，也就是从<K1,V1>到<K2,V2>到<K3,V3>的一个过程。
Map端
<1>在map端首先接触的是InputSplit,在InputSplit中含有DataNode中的数据,每个InputSplit都会分配一个Mapper任务,Mapper任务结束后产生<K2,V 2>的输出，这些输出显存放在缓存中，每一个map有一个环形内存缓冲区，用于存储任务的输出。默认大小100MB（io.sort.mb属性），一旦达到阀值0.8(io.sort.spil l.percent),一个后台线程就把内容写到(spill)Linux本地磁盘中的指定目录（mapred.local.dir）下的新建的一个溢出写文件。
<2>写磁盘前，要partition,sort。经过分区，将不一样类型的数据分开处理，以后对不一样分区的数据进行排序，若是有Combiner，还要对排序后的数据进行co mbine。等最后记录写完，将所有溢出文件合并为一个分区且排序的文件。
<3>最后将磁盘中的数据送到Reduce中，从图中能够看出Map输出有三个分区，有一个分区数据被送到图示的Reduce任务中，剩下的两个分区被送到其余Reducer任务中。而图示的Reducer任务的其余的三个输入来自其余的Map输出。
Reduce端
<1>Reducer经过Http方式获得输出文件的分区。
<2>TaskTracker为分区文件运行Reduce任务。复制阶段把Map输出复制到Reducer的内存或磁盘。一个Map任务完成，Reduce就开始复制输出。
<3>排序阶段合并map输出。而后走Reduce阶段。
3.2 Hadoop压缩算法
3.2.1 算法介绍
Hadoop的压缩过程并非一个必须的过程，但为何还要使用呢？在哪些阶段可使用，有什么好处呢？
<1>在Map输出到Reduce时可使用，由于map端输出的数据要经过网络输出到Reduce端，为了减小传输的数据量咱们能够采用压缩的方式来减小延迟。
<2>在整个做业的输出也可使用
Codec为是压缩，解压缩的算法的实现，在Hadoop中，codec由CompressionCode的实现来表示。下面是一些实现，以下图3.1所示。
图 3.1
3.2.2 MapReduce的输出进行压缩
输出的压缩属性，和使用方式：以下图3.2,3.3所示。
图 3.2
图3.3
3.3 常见算法
3.3.1 MapReduce常见算介绍
<1>单词计数(已介绍)
<2>数据去重(去掉重复数据不难理解吧)
<3>排序(在上节已介绍)
<4>Top K(是求最值问题，下面会介绍)
下面算法，跟咱们数据库中的方法比较相似，
<5>选择---行
数据库中：该操做的结果应该是一行一行的显示，至关于where。
MapReduce的实现：以求最值为例，从100万数据中选出一行最小值。
<6>投影---列
数据库中：该操做的结果应该是一列一列的显示，至关于select。
MapReduce的实现：以求处理手机上网日志为例，从其11个字段选出了五个字段来显示咱们的手机上网流量。
<7>分组
数据库中：至关于group by。
MapReduce的实现：至关于分区，以求处理手机上网日志为例，喊手机号和非手机号分为两组。
<9>多表链接
MapReduce中：在MapReduce中能够同时进入多个文件进行操做，其中两个文件有关系就至关于表链接。那么如何知道文件之间的关系呢？我能够经过map函数的context参数来得到文件路径代码以下
final FileSplit inputSplit = (FileSplit) context.getInputSplit(); final String path = inputSplit.getPath().toString();
<10>单表关联
经过上面的分析咱们能够知道，sql中的方法也能够在MapReduce中实现，也就是说当把关系型数据库中的算法所有在MapReduce中实现时，也就意味着sql的使用范围扩展到了Hadoop，也就是大数据领域，这样意义是很是大的。
3.3.2 Top K 最值案例
求最值的方法，在咱们的生活中应用很是的广，好比找出高考中的最高分，也就是状元，就很是相似分布式计算，要选出全国的最高分就首先选出各省份的，要选出各省份就得选出各市级的等等，而这些数据量很是大，没法直接所有加载到内存中，面对如此大的数据量我就能够考虑使用分布式计算的方式。咱们以从100万的数据中求出其中的最大值为例，介绍该方法。
求最值最简单的办法就是对该文件进行一次遍历得出最值，可是现实中数据比量比较大，这种方法不能实现。在传统的MapReduce思想中，将文件的数据通过map迭代出来送到reduce中，在Reduce中求出最大值。但这个方法显然不够优化，咱们可采用“分而治之”的思想，不须要map的全部数据所有送到reduce中，咱们能够在map中先求出最大值，将该map任务的最大值送reduce中，这样就减小了数据的传输量。那么何时该把这个数据写出去呢？咱们知道，每个键值对都会调用一次map()，因为数据量大调用map()的次数也就多了，显然在map()函数中将该数据写出去是不明智的，因此最好的办法该Mapper任务结束后将该数据写出去。咱们又知道，当Mapper/Reducer任务结束后会调用cleanup函数，因此咱们能够在该函数中将该数据写出去。了解了这些咱们能够看一下程序的代码如3.1所示。
1 package suanfa; 2 3 import java.net.URI; 4 5 import mapreduce.WordCountApp; 6 7 import org.apache.hadoop.conf.Configuration; 8 import org.apache.hadoop.fs.FileSystem; 9 import org.apache.hadoop.fs.Path; 10 import org.apache.hadoop.io.LongWritable; 11 import org.apache.hadoop.io.NullWritable; 12 import org.apache.hadoop.io.Text; 13 import org.apache.hadoop.mapreduce.Job; 14 import org.apache.hadoop.mapreduce.Mapper; 15 import org.apache.hadoop.mapreduce.Reducer; 16 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 17 import org.apache.hadoop.mapreduce.lib.input.FileSplit; 18 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 19 20 public class TopKApp { 21 static final String INPUT_PATH = "hdfs://hadoop:9000/input2"; 22 static final String OUT_PATH = "hdfs://hadoop:9000/out2"; 23 24 public static void main(String[] args) throws Exception { 25 Configuration conf = new Configuration(); 26 final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf); 27 final Path outPath = new Path(OUT_PATH); 28 if(fileSystem.exists(outPath)){ 29 fileSystem.delete(outPath, true); 30 } 31 32 final Job job = new Job(conf , WordCountApp.class.getSimpleName()); 33 FileInputFormat.setInputPaths(job, INPUT_PATH); 34 job.setMapperClass(MyMapper.class); 35 job.setReducerClass(MyReducer.class); 36 job.setOutputKeyClass(LongWritable.class); 37 job.setOutputValueClass(NullWritable.class); 38 FileOutputFormat.setOutputPath(job, outPath); 39 job.waitForCompletion(true); 40 } 41 static class MyMapper extends Mapper<LongWritable, Text, LongWritable, NullWritable>{ 42 long max = Long.MIN_VALUE; 43 protected void map(LongWritable k1, Text v1, Context context) throws java.io.IOException ,InterruptedException { 44 final long temp = Long.parseLong(v1.toString()); 45 if(temp>max){ 46 max = temp; 47 } 48 }; 49 50 protected void cleanup(org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,LongWritable, NullWritable>.Context context) throws java.io.IOException ,InterruptedException { 51 context.write(new LongWritable(max), NullWritable.get()); 52 }; 53 } 54 55 static class MyReducer extends Reducer<LongWritable, NullWritable, LongWritable, NullWritable>{ 56 long max = Long.MIN_VALUE; 57 protected void reduce(LongWritable k2, java.lang.Iterable<NullWritable> arg1, org.apache.hadoop.mapreduce.Reducer<LongWritable,NullWritable,LongWritable,NullWritable>.Context arg2) throws java.io.IOException ,InterruptedException { 58 final long temp = k2.get(); 59 if(temp>max){ 60 max = temp; 61 } 62 }; 63 64 protected void cleanup(org.apache.hadoop.mapreduce.Reducer<LongWritable,NullWritable,LongWritable,NullWritable>.Context context) throws java.io.IOException ,InterruptedException { 65 context.write(new LongWritable(max), NullWritable.get()); 66 }; 67 } 68 }
代码3.1
运行结果为：32767，也就是咱们数据中的最大值
相关文章
- 1. hadoop自定义排序，分组排序
- 2. hadoop-之二次排序&分组&分区
- 3. hadoop 的排序：key排序和分组排序
- 4. sql分组排序笔记
- 5. Hadoop之MapReduce中的排序和分组（笔记11）
- 6. Python Hadoop Mapreduce 实现Hadoop Streaming分组和二次排序
- 7. Hadoop-Python实现Hadoop Streaming分组和二次排序
- 8. python算法日记（归并排序）_leetcode 148. 排序链表 912. 排序数组
- 9. Sqlserver 分组排序，排名
- 10. Mysql 分组排序
- 更多相关文章...
- • PHP 数组排序 - PHP教程
- • ADO 排序 - ADO 教程
- • 算法总结-归并排序
- • Tomcat学习笔记（史上最全tomcat学习笔记）
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章