设为首页 加入收藏

TOP

使用Hadoop MapReduce 进行排序(一)
2014-11-24 07:15:02 来源: 作者: 【 】 浏览:0
Tags:使用 Hadoop MapReduce 进行 排序

排序的基本思想是利用了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函数,来定义比较的规则,从而可以实现字符串或其他非数字类型的排序,也可以实现二次排序乃至多次排序。


CxfInputFormat.java

package com.alibaba.cxf.sort;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.LineRecordReader;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;

public class CxfInputFormat extends FileInputFormat{
@Override
public RecordReader getRecordReader(InputSplit split,
JobConf job, Reporter reporter) throws IOException {
return new CxfRecordReader(job, (FileSplit) split);
}
class CxfRecordReader implements RecordReader {

private LineRecordReader in;
private LongWritable junk = new LongWritable();
private Text line = new Text();
private int KEY_LENGTH = 10;
public CxfRecordReader(JobConf job,FileSplit split) throws IOException{
in = new LineRecordReader(job, split);
}
@Override
public void close() throws IOException {
in.close();
}
@Override
public IntWritable createKey() {
return new IntWritable();
}
@Override
public Text createva lue() {

return new Text();
}
@Override
public long getPos() throws IOException {

return in.getPos();
}
@Override
public float getProgress() throws IOException {

return in.getProgress();
}
@Override
public boolean next(IntWritable key, Text value) throws IOException {
if (in.next(junk, line)) {
if (line.getLength() < KEY_LENGTH) {
key.set(Integer.parseInt(line.toString()));
value = new Text();
// value.clear();
} else {
byte[] bytes = line.getBytes();
key.set(Integer.parseInt(new String(bytes).substring(0, KEY_LENGTH)));
value = new Text();
}
return true;
} else {
return false;
}
}
}
}



SortByMapReduce.java

package com.alibaba.cxf.sort;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.

首页 上一页 1 2 下一页 尾页 1/2/2
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
分享到: 
上一篇自定义Hadoop Map/Reduce输入文件.. 下一篇Hadoop涉及GBK编码的文件

评论

帐  号: 密码: (新用户注册)
验 证 码:
表  情:
内  容:

·MySQL 安装及连接-腾 (2025-12-25 06:20:28)
·MySQL的下载、安装、 (2025-12-25 06:20:26)
·MySQL 中文网:探索 (2025-12-25 06:20:23)
·Shell脚本:Linux Sh (2025-12-25 05:50:11)
·VMware虚拟机安装Lin (2025-12-25 05:50:08)