`

使用Hadoop MapReduce 进行排序

 
阅读更多

本文转自:http://www.alidw.com/?p=1420

在hadoop中的例子TeraSort,就是一个利用mapredue进行排序的例子。本文参考并简化了这个例子:

排序的基本思想是利用了mapreduce的自动排序功能,在hadoop中,从map到reduce阶段,map出来的结构会按照各个key按照 hash值分配到各个reduce中,其中,在reduce中所有的key都是有序的了。如果使用一个reduce,那么我们直接将他output出来就 行了,但是这不能够体现分布式的好处,所以,我们还是要用多个reduce来跑。

比方说我们有1000个1-10000的数据,跑10个ruduce任务, 如果我们运行进行partition的时候,能够将在1-1000中数据的分配到第一个reduce中,1001-2000的数据分配到第二个 reduce中,以此类推。即第n个reduce所分配到的数据全部大于第n-1个reduce中的数据。这样,每个reduce出来之后都是有序的了, 我们只要cat所有的输出文件,变成一个大的文件,就都是有序的了

基本思路就是这样,但是现在有一个问题,就是数据的区间如何划分,在数据量大,还有我们并不清楚数据分布的情况下。一个比较简单的方法就是采样,假如有一 亿的数据,我们可以对数据进行采样,如取10000个数据采样,然后对采样数据分区间。在Hadoop中,patition我们可以用 TotalOrderPartitioner替换默认的分区。然后将采样的结果传给他,就可以实现我们想要的分区。在采样时,我们可以使用hadoop的 几种采样工具,RandomSampler,InputSampler,IntervalSampler。

这样,我们就可以对利用分布式文件系统进行大数据量的排序了,我们也可以重写Partitioner类中的compare函数,来定义比较的规则,从而可以实现字符串或其他非数字类型的排序,也可以实现二次排序乃至多次排序。

参考:《Hadoop权威指南》里面有详细的讲

复制代码
  1 CxfInputFormat.java
  2 
  3 package com.alibaba.cxf.sort;
  4 
  5 import java.io.IOException;
  6 
  7 import org.apache.hadoop.io.IntWritable;
  8 import org.apache.hadoop.io.LongWritable;
  9 import org.apache.hadoop.io.NullWritable;
 10 import org.apache.hadoop.io.Text;
 11 import org.apache.hadoop.mapred.FileInputFormat;
 12 import org.apache.hadoop.mapred.FileSplit;
 13 import org.apache.hadoop.mapred.InputSplit;
 14 import org.apache.hadoop.mapred.JobConf;
 15 import org.apache.hadoop.mapred.LineRecordReader;
 16 import org.apache.hadoop.mapred.RecordReader;
 17 import org.apache.hadoop.mapred.Reporter;
 18 
 19 public class CxfInputFormat extends FileInputFormat<IntWritable,Text>{
 20  @Override
 21  public RecordReader<IntWritable, Text> getRecordReader(InputSplit split,
 22    JobConf job, Reporter reporter) throws IOException {
 23   return new CxfRecordReader(job, (FileSplit) split);
 24  }
 25  class CxfRecordReader implements RecordReader<IntWritable,Text> {
 26 
 27   private LineRecordReader in;
 28      private LongWritable junk = new LongWritable();
 29      private Text line = new Text();
 30      private  int KEY_LENGTH = 10;
 31   public CxfRecordReader(JobConf job,FileSplit split) throws IOException{
 32    in = new LineRecordReader(job, split);
 33   }
 34   @Override
 35   public void close() throws IOException {
 36    in.close();   
 37   }
 38   @Override
 39   public IntWritable createKey() {
 40    return new IntWritable();
 41   }
 42   @Override
 43   public Text createValue() {
 44    
 45    return new Text();
 46   }
 47   @Override
 48   public long getPos() throws IOException {
 49    
 50    return in.getPos();
 51   }
 52   @Override
 53   public float getProgress() throws IOException {
 54    
 55    return in.getProgress();
 56   }
 57   @Override
 58   public boolean next(IntWritable key, Text value) throws IOException {
 59    if (in.next(junk, line)) {
 60     if (line.getLength() < KEY_LENGTH) {
 61      key.set(Integer.parseInt(line.toString()));
 62      value = new Text();
 63   //   value.clear();
 64     } else {
 65      byte[] bytes = line.getBytes();
 66      key.set(Integer.parseInt(new String(bytes).substring(0, KEY_LENGTH)));
 67      value = new Text();
 68     }
 69     return true;
 70    } else {
 71     return false;
 72    }
 73   }
 74  }
 75 }
 76 
 77  
 78 
 79 SortByMapReduce.java
 80 
 81 package com.alibaba.cxf.sort;
 82 
 83 import java.io.IOException;
 84 import java.net.URI;
 85 import java.net.URISyntaxException;
 86 import org.apache.hadoop.filecache.DistributedCache;
 87 import org.apache.hadoop.fs.Path;
 88 import org.apache.hadoop.io.IntWritable;
 89 import org.apache.hadoop.io.NullWritable;
 90 import org.apache.hadoop.mapred.FileInputFormat;
 91 import org.apache.hadoop.mapred.FileOutputFormat;
 92 import org.apache.hadoop.mapred.JobClient;
 93 import org.apache.hadoop.mapred.JobConf;
 94 import org.apache.hadoop.mapred.TextOutputFormat;
 95 import org.apache.hadoop.mapred.lib.InputSampler;
 96 import org.apache.hadoop.mapred.lib.TotalOrderPartitioner;
 97 public class SortByMapReduce {
 98 
 99  /**
100   * @param args
101   * @throws URISyntaxException
102   * @throws IOException
103 */
104  public static void main(String[] args) throws IOException, URISyntaxException {
105   runJob(args);
106  }
107 
108  private static void runJob(String[] args) throws IOException, URISyntaxException {
109   
110   JobConf conf = new JobConf(SortByMapReduce.class);
111   
112   FileInputFormat.setInputPaths(conf, new Path(args[0]));
113         FileOutputFormat.setOutputPath(conf, new Path(args[1]));
114         conf.setJobName(”SortByMapReduce”);
115   
116   conf.setInputFormat(CxfInputFormat.class);
117   conf.setOutputKeyClass(IntWritable.class);
118   conf.setOutputFormat(TextOutputFormat.class);
119   conf.setNumReduceTasks(5);
120   conf.setPartitionerClass(TotalOrderPartitioner.class);
121   InputSampler.RandomSampler<IntWritable, NullWritable> sampler =
122    new InputSampler.RandomSampler<IntWritable, NullWritable>(0.1,10000,10);
123   
124   Path input = FileInputFormat.getInputPaths(conf)[0];
125   input = input.makeQualified(input.getFileSystem(conf));
126   Path partitionFile = new Path(input,”_partitions”);
127   TotalOrderPartitioner.setPartitionFile(conf, partitionFile);
128   InputSampler.writePartitionFile(conf, sampler);
129   
130   URI partitionURI = new URI(partitionFile.toString() + “#_partitions”);
131   DistributedCache.addCacheFile(partitionURI, conf);
132   DistributedCache.createSymlink(conf);
133   JobClient.runJob(conf);  
134  } 
135 }
复制代码
分享到:
评论

相关推荐

    大数据 hadoop mapreduce 词频统计

    在hadoop平台上,用mapreduce编程实现大数据的词频统计

    hadoop 二次排序 原理

    Hadoop 大数据方向 mapreduce计算中的二次排序,讲解透彻

    mapreduce二次排序

    mapreduce二次排序,年份升序,按照年份聚合,气温降序

    Hadoop MapReduce高级特性

    本文针对MapReduce中的包括计数器、排序和数据集连接进行讲解。计数器包括系统计数器和自定义计数器,排序包括部分排序和文件排序,还包括对键内数据的排序。

    mapreduce实现全栈排序

    mapreduce实现全栈排序,简单算法已经在文档中说明,想要了解的可以查看!

    Hadoop大作业排序.zip

    由于 MapReduce 中对 key 进行比较和排序,而 key 可以是任何实 现了 Writable 接口的类。 在 java 中,要实现类的大小比较可以实现 Comparable 接口并通 过重写 compareTo 方法来实现。 在 Mapreduce 中,如果需要...

    Hadoop中MapReduce基本案例及代码(五)

    对相同分区的数据,按照key进行排序(默认按照字典排序)、分组。相同key的value放在一个集合中。 (可选)分组后对数据进行归约。 注意:MapReduce中,Mapper可以单独存在,但是Reducer不能存在。

    hadoop 1.2.1 api 最新chm 伪中文版

    框架会对map的输出先进行排序, 然后把结果输入给reduce任务。通常作业的输入和输出都会被存储在文件系统中。 整个框架负责任务的调度和监控,以及重新执行已经失败的任务。 通常,hadoop Map/Reduce框架和分布式...

    UDA:Hadoop MapReduce 的非结构化数据加速器 (RDMA)

    一种新颖的数据移动协议将 RDMA 与高效的归并排序算法结合使用,使基于 InfiniBand 和 10GbE 和 40GbE RoCE(RDMA over Converged Ethernet)适配器卡的 Hadoop 集群能够在加速 Hadoop 框架的服务器之间有效移动...

    Hadoop Shuffle过程全解析

    Hadoop Mapreduce过程shuffle过程全解析,Shuffle过程

    mapreduce案例代码及案例涉及文件

    了解map和reduce工作原理,以及排序,分组,分区设置,有详细的注释,方便查看学习,适合入门初学者练手

    基于MapReduce的简单倒排索引的建立

    基于MapReduce的简单倒排索引的建立

    Hadoop从入门到上手企业开发

    060 MapReduce执行流程之Shuffle和排序流程以及Map端分析 061 MapReduce执行流程之Reduce端分析 062 MapReduce Shuffle过程讲解和Map Shuffle Phase讲解 063 Reduce Shuffle Phase讲解 064 源代码跟踪查看Map Task和...

    基于ItemCF协同过滤、hadoop-mapreduce的商品推荐系统下载地址

    基于ItemCF协同过滤、hadoop-mapreduce的商品推荐系统下载地址。 基于ItemCF的协同过滤 物品推荐系统 Collaborative filtering goods recommendation system based on ItemCF Step1.run(config, paths); // 格式化 ...

    论文研究-基于Hadoop的多关键字排序方法研究.pdf

    在单机环境下按多关键字对大数据排序需要较长的执行时间,为了提高按多关键字对大数据排序的效率,根据Hadoop的MapReduce模型,给出了两种基于Hadoop的多关键字排序方法。方法一在Reduce函数中使用链式基数排序算法...

    hadoop实现计数器

    hadoop实现计数器,利用mapreduce进行排序,同时可以实时跟踪整个过程以及每个节点的参与情况

    使用Java MapReduce实现数据全局排序【100012685】

    本次实验,在 Hadoop 平台上,使用 MapReduce 实现了数据的全局排序。将详细阐述了实现所需环境及过程。用阿里云服务器安装, OS: Ubuntu20.04 LTS . Hadoop 支持用三种模式启动:单机模式、伪分布式模式、分布式...

    Hadoop权威指南 第二版(中文版)

    Hadoop的I/O、MapReduce应用程序开发;MapReduce的工作机制;MapReduce的类型和格式;MapReduce的特性;如何构建Hadoop集群,如何管理Hadoop;Pig简介;Hbase简介;Hive简介;ZooKeeper简介;开源工具Sqoop,最后还...

    新版Hadoop视频教程 段海涛老师Hadoop八天完全攻克Hadoop视频教程 Hadoop开发

    04-hadoop的自定义排序实现.avi 05-mr程序中自定义分组的实现.avi 06-shuffle机制.avi 07-mr程序的组件全貌.avi 08-textinputformat对切片规划的源码分析.avi 09-倒排索引的mr实现.avi 10-多个job在同一个...

    word源码java-hadoop-test:hadoop、mapreduce的一些练习

    包org.dan.mr.flowsumsort MapReduce流量统计,按总流量排序 包org.dan.mr.order_pro MapReduce实现订单信息和产品信息的join逻辑 包org.dan.mr.order_pro_mapjoin MapReduce实现订单信息和产品信息的join逻辑,在...

Global site tag (gtag.js) - Google Analytics