图文精华

About云-梭伦科技»专题 › 交流区 › 技术交流 › MapReduce › 【转】【mapreduce进阶编程五】全局排序
分享

【转】【mapreduce进阶编程五】全局排序

cknote 发表于 2014-9-30 11:18:06 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 3 21120
About云VIP帮助找工作和提升


Hadoop平台没有提供全局数据排序，而在大规模数据处理中进行数据的全局排序是非常普遍的需求。本文首先实现简单的全局排序，然后再使用Hadoop平台提供的采样器解决数据倾斜的问题。
0.环境：
   hadoop2.4.0 四节点：master(192.168.137.100) slave1(192.168.137.101) slave2(192.168.137.102) slave3(192.168.137.103)
window8下通过eclipse的插件远程提交job到hadoop环境

1.描述：
1.1数据输入：

  1. 5

  2. 7

  3. 6

  4. 9

  5. 1

  6. 11

  7. 79

  8. 36

  9. 55

  10. 66

  11. 77

  12. 34

  13. 29

  14. 99

  15. 101

  16. 202

  17. 987

  18. 123

  19. 456

  20. 714

  21. 562

  22. 234

  23. 1234

  24. 611

  25. 3

  26. 97

  27. 82
复制代码


1.2预计输出：

全局排序，见运行截图

2.设计思路：

   我们可以选择reduce作为节点来分割数据，例如把小于100的让一个ruduce处理，大于100小于1000的第二个reduce处理。。。，每个reduce内部都是排好序的，就实现了全局排序；但是这样硬分割的后果是会产生数据倾斜，分布不均匀，可能会大部分的数据都在某个reduce上，或者某个reduce上的数据非常稀疏。hadoop平台给我们提供了采样器，可以解决这个问题，我们使用简单的随机采样。

3.代码实现：
3.1编写mapper

  1. package com.cknote.hadoop.globlesort;

  2. 

  3. import java.io.IOException;

  4. 

  5. import org.apache.hadoop.io.LongWritable;

  6. import org.apache.hadoop.io.NullWritable;

  7. import org.apache.hadoop.io.Text;

  8. import org.apache.hadoop.mapreduce.Mapper;

  9. 

  10. public class MyMapper extends

  11.                 Mapper<LongWritable, Text, LongWritable, NullWritable> {

  12. 

  13.         @Override

  14.         protected void map(LongWritable key, Text value, Context context)

  15.                         throws IOException, InterruptedException {

  16. 

  17.                 context.write(new LongWritable(Integer.parseInt(value.toString())),

  18.                                 NullWritable.get());

  19. 

  20.         }

  21. 

  22. }
复制代码


3.2编写reducer


  1. package com.cknote.hadoop.globlesort;

  2. 

  3. import java.io.IOException;

  4. 

  5. import org.apache.hadoop.io.LongWritable;

  6. import org.apache.hadoop.io.NullWritable;

  7. import org.apache.hadoop.mapreduce.Reducer;

  8. 

  9. public class MyReducer extends

  10.                 Reducer<LongWritable, NullWritable, LongWritable, NullWritable> {

  11. 

  12.         @Override

  13.         protected void reduce(LongWritable key, Iterable<NullWritable> value,

  14.                         Context context) throws IOException, InterruptedException {

  15. 

  16.                 context.write(key, NullWritable.get());

  17. 

  18.         }

  19. 

  20. }
复制代码




3.3编写分区

  1. package com.cknote.hadoop.globlesort;

  2. 

  3. import org.apache.hadoop.io.LongWritable;

  4. import org.apache.hadoop.io.NullWritable;

  5. import org.apache.hadoop.mapreduce.Partitioner;

  6. 

  7. public class MyPartitioner extends Partitioner<LongWritable, NullWritable> {

  8. 

  9.         @Override

  10.         public int getPartition(LongWritable key, NullWritable value,

  11.                         int numPartitions) {

  12.                 long tmp = key.get();

  13.                 if (tmp <= 100) {

  14.                         return 0 % numPartitions;

  15.                 } else if (tmp <= 1000) {

  16.                         return 1 % numPartitions;

  17.                 } else {

  18.                         return 2 % numPartitions;

  19.                 }

  20.         }

  21. 

  22. }
复制代码





3.4编写主类
  1. package com.cknote.hadoop.globlesort;

  2. 

  3. import org.apache.hadoop.conf.Configuration;

  4. import org.apache.hadoop.fs.Path;

  5. import org.apache.hadoop.io.LongWritable;

  6. import org.apache.hadoop.io.NullWritable;

  7. import org.apache.hadoop.mapreduce.Job;

  8. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

  9. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

  10. import org.apache.hadoop.util.GenericOptionsParser;

  11. 

  12. public class GlobleSortMain1 {

  13. 

  14.         public static void main(String[] args) throws Exception {

  15.                 Configuration conf = new Configuration();

  16.                 String[] tmpArgs = {

  17.                                 "hdfs://192.168.137.100:9000/user/hadoop/globlesort/in",

  18.                                 "hdfs://192.168.137.100:9000/user/hadoop/globlesort/out" };

  19. 

  20.                 String[] otherArgs = null;

  21.                 if (args.length != 3) {

  22.                         otherArgs = new GenericOptionsParser(conf, tmpArgs)

  23.                                         .getRemainingArgs();

  24.                 } else {

  25.                         otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

  26.                 }

  27. 

  28.                 Job job = new Job(conf, "globle sort 1");

  29. 

  30.                 job.setJarByClass(GlobleSortMain1.class);

  31.                 job.setMapperClass(MyMapper.class);

  32.                 job.setReducerClass(MyReducer.class);

  33.                 job.setPartitionerClass(MyPartitioner.class);

  34. 

  35.                 job.setNumReduceTasks(3);

  36. 

  37.                 job.setMapOutputKeyClass(LongWritable.class);

  38.                 job.setMapOutputValueClass(NullWritable.class);

  39.                 job.setOutputKeyClass(LongWritable.class);

  40.                 job.setOutputValueClass(NullWritable.class);

  41. 

  42.                 FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

  43.                 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

  44.                 System.exit(job.waitForCompletion(true) ? 0 : 1);

  45. 

  46.         }

  47. }
复制代码



4.运行结果：
1.jpg

2.jpg

3.jpg

5.代码实现
5.1编写主类
   mapper和reducer无须重新编写

  1. package com.cknote.hadoop.globlesort;

  2. 

  3. import java.net.URI;

  4. import java.util.Date;

  5. 

  6. import org.apache.hadoop.conf.Configuration;

  7. import org.apache.hadoop.fs.Path;

  8. import org.apache.hadoop.io.IntWritable;

  9. import org.apache.hadoop.io.LongWritable;

  10. import org.apache.hadoop.io.NullWritable;

  11. import org.apache.hadoop.mapreduce.Job;

  12. import org.apache.hadoop.mapreduce.filecache.DistributedCache;

  13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

  14. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

  15. import org.apache.hadoop.mapreduce.lib.partition.InputSampler;

  16. import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;

  17. import org.apache.hadoop.util.GenericOptionsParser;

  18. 

  19. public class GlobleSortMain2 {

  20. 

  21.         public static void main(String[] args) throws Exception {

  22.                 Configuration conf = new Configuration();

  23.                 String[] tmpArgs = {

  24.                                 "hdfs://192.168.137.100:9000/user/hadoop/globlesort/in",

  25.                                 "hdfs://192.168.137.100:9000/user/hadoop/globlesort/out2" };

  26. 

  27.                 String[] otherArgs = null;

  28.                 if (args.length != 2) {

  29.                         otherArgs = new GenericOptionsParser(conf, tmpArgs)

  30.                                         .getRemainingArgs();

  31.                 } else {

  32.                         otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

  33.                 }

  34. 

  35.                 Job job = new Job(conf, "globle sort 2");

  36. 

  37.                 job.setJarByClass(GlobleSortMain2.class);

  38.                 job.setMapperClass(MyMapper.class);

  39.                 job.setReducerClass(MyReducer.class);

  40. 

  41.                 job.setNumReduceTasks(3);

  42. 

  43.                 job.setMapOutputKeyClass(LongWritable.class);

  44.                 job.setMapOutputValueClass(NullWritable.class);

  45.                 job.setOutputKeyClass(LongWritable.class);

  46.                 job.setOutputValueClass(NullWritable.class);

  47. 

  48.                 FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

  49.                 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

  50. 

  51.                 job.setPartitionerClass(TotalOrderPartitioner.class);

  52.                 InputSampler.RandomSampler<LongWritable, NullWritable> sampler 

  53.                 = new InputSampler.RandomSampler<LongWritable, NullWritable>(

  54.                                 0.1, 100, 3);

  55. 

  56.                 Path partitionFile = new Path(otherArgs[0], "_sortpartitions");

  57. 

  58.                 TotalOrderPartitioner.setPartitionFile(conf, partitionFile);

  59. 

  60.                 InputSampler.writePartitionFile(job, sampler);

  61. 

  62.                 // 一般都将该文件做distribute cache处理

  63.                 URI partitionURI = new URI(partitionFile.toString()

  64.                                 + "#_sortpartitions");

  65.                 // 从上面可以看出 采样器是在map阶段之前进行的

  66.                 // 在提交job的client端完成的

  67.                 DistributedCache.addCacheFile(partitionURI, conf);

  68.                 DistributedCache.createSymlink(conf);

  69. 

  70.                 Date startTime = new Date();

  71.                 System.out.println("Job started: " + startTime);

  72.                 System.exit(job.waitForCompletion(true) ? 0 : 1);

  73.                 Date end_time = new Date();

  74.                 System.out.println("Job ended: " + end_time);

  75.                 System.out.println("The job took "

  76.                                 + (end_time.getTime() - startTime.getTime()) / 1000

  77.                                 + " seconds.");

  78. 

  79.         }

  80. }
复制代码





6运行结果

4.jpg
5.jpg

6.jpg





回复

使用道具 举报

自定义广告语

已有(3)人评论

跳转到指定楼层
正序浏览
sun1 发表于 2015-4-22 07:49:09
感谢分享，学习了。
回复

使用道具 举报

admln 发表于 2015-9-23 15:51:39
The type DistributedCache is deprecated
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
cknote

新手上路

关注

  • 2

    主题

  • 12

    帖子

  • 3

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000