MapReduce对大数据进行排序的实践

从一个小的例子开始:

Map和reduce之间的shuffle(洗牌)是将key值相同的放在一块,sort(排序)是按照key值进行排序.例如like出现了两次,就会把like放在一起.you也是.然后根据key值进行按照字典的顺序进行排序.下面我想将下面的数据按照时间进行排序,并且ID相同的在一块.具体的数据格式:(数据简化成如下,其实还包含其他的一些数据)

6395        1473840570 

6393        1473840390 

6393        1473840150

6393        1473840450

6395        1473840030 

6395        1473840991 

6394        1473839970 

6394        1473840811 

6394        1473840090

......................

在这里我还是要推荐下我自己建的大数据学习交流群：532218147，群里都是学大数据开发的，如果你正在学习大数据 ，小编欢迎你加入，大家都是软件开发党，不定期分享干货（只有大数据开发相关的），包括我自己整理的一份2018最新的大数据进阶资料和高级开发教程，欢迎进阶中和进想深入大数据的小伙伴。

第一列是ID号,第二列是Linux时间戳.想输出的结果是:ID号相同的放在一块,对应的Linux时间戳从小到大进行排序.在编写代码之前可以想到,底层的MapReduce已经帮我们做了一些工作:对key值相同进行聚集(shuffle洗牌).这里是看主要是reduce部分,map部分工作已经简化了.

刚开始的时候,我是这么想的:

map(k0,v0)-->list(k1,v1)

key: k1 是ID号, Value: v1 是时间戳

reduce(k1,list(v1))  -->list(k2,v2)

在reduce中对list(v1),也就是时间戳列表,先添加到list中,然后用Collections的sort方法对list进行排序,最后将结果进行输出.程序如下:

publicclassSortReducerextendsReducer{@Overrideprotectedvoidreduce(Text key, Iterable values,

              Context context)throwsIOException, InterruptedException{// TODO Auto-generated method stubList queue =newArrayList();for(LongWritable num: values) {              queue.add(num);        }        Collections.sort(queue);//排序//将排序结果输出for(LongWritable num:queue) {              context.write(key, num);        }    }}

貌似逻辑没有一点问题,可是结果却是:

所有的时间都是一样的.这是为什么呢?于是网上百度一番,终于找到原因了:

reduce方法会反复执行多次，但key和value相关的对象只有两个，reduce会反复重用这两个对象。所以如果要保存key或者value的结果，只能将其中的值取出另存或者重新clone一个对象（例如Text store = new Text(value) 或者 String a = value.toString()），而不能直接赋引用。因为引用从始至终都是指向同一个对象，你如果直接保存它们，那最后它们都指向最后一个输入记录。会影响最终计算结果而出错。

解决方案:

String str = num.toString();

LongWritable num1 = new LongWritable(Long.parseLong(str));

修改后的源代码如下:

publicclassSortReducerextendsReducer{@Overrideprotectedvoidreduce(Text key, Iterable values,

              Context context)throwsIOException, InterruptedException{// TODO Auto-generated method stubList queue =newArrayList();for(LongWritable num: values) {              String str = num.toString();              LongWritable num1 =newLongWritable(Long.parseLong(str));              queue.add(num1);        }        Collections.sort(queue);for(LongWritable num:queue) {              String str = num.toString();              LongWritable num1 =newLongWritable(Long.parseLong(str));              context.write(key, num1);        }    }}

最终结果:

后文

这是我在大数据学习的个人总结,如果有错误的地方或者还有改进的地方希望大家不吝赐教.，更多IT编程分享请关注微信公众号：程序员大牛，每天分享干货！