hadoop mapreduce排序原理，hadoopmapreduce

hadoop mapreduce排序原理，hadoopmapreduce

                                                           hadoop  mapreduce排序原理

Hadoop 案例3----数据排序  简单问题  （入门级别）


"数据排序"是许多实际任务执行时要完成的第一项工作，
比如学生成绩评比、数据建立索引等。这个实例和数据去重类似，都是先对原始数据进行初步处理，为进一步的数据操作打好基础。下面进入这个示例。


1、需求描述
 对输入文件中数据进行排序。输入文件中的每行内容均为一个数字，即一个数据。
 要求在输出中每行有两个间隔的数字，其中，第一个代表原始数据在原始数据集中的位次，第二个代表原始数据。


2、原始数据


 1）file1：


2


32


654


32


15


756


65223


2）file2：
5956


22


650


92


3）file3：


26


54


6


 
    样例输出：


    1    2


    2    6


    3    15


    4    22


    5    26


    6    32


    7    32


    8    54


    9    92


    10    650


    11    654


    12    756


    13    5956


    14    65223


3、设计思考
这个实例仅仅要求对输入数据进行排序，熟悉MapReduce过程的读者会很快想到在MapReduce过程中就有排序，是否可以利用这个默认的排序，
而不需要自己再实现具体的排序呢？答案是肯定的。


但是在使用之前首先需要了解它的默认排序规则。它是按照key值进行排序的，如果key为封装int的IntWritable类型，那么MapReduce按照数字大小对key排序，
如果key为封装为String的Text类型，那么MapReduce按照字典顺序对字符串排序。


了解了这个细节，我们就知道应该使用封装int的IntWritable型数据结构了。也就是在map中将读入的数据转化成 IntWritable型，然后作为key值输出（value任意）。
reduce拿到<key，value-list>之后，将输入的 key作为value输出，并根据value-list中元素的个数决定输出的次数。
输出的key（即代码中的linenum）是一个全局变量，它统计当前key的位次。需要注意的是这个程序中没有配置Combiner，也就是在MapReduce过程中不使用Combiner。
这主要是因为使用map和reduce就已经能够完成任务了。


4、map代码
package com.wy.hadoop.sort;


import java.io.IOException;


import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;


public class IntSortMapper extends Mapper<Object, Text, IntWritable, IntWritable> {


IntWritable val = new IntWritable(1);
@Override
protected void map(Object key, Text value,Context context)
throws IOException, InterruptedException {

context.write(new IntWritable(Integer.valueOf(value.toString())), val);

}



}


5、reduce代码
package com.wy.hadoop.sort;


import java.io.IOException;


import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;


public class IntSortReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {


IntWritable num = new IntWritable(1);
@Override
protected void reduce(IntWritable key, Iterable<IntWritable> values,Context context)
throws IOException, InterruptedException {
for(IntWritable tmp: values){
context.write(num, tmp);
num = new IntWritable(num.get()+1);
}
}

}




6、main代码
package com.wy.hadoop.sort;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


import com.wy.hadoop.join.two.UserJob;


public class IntSortJob extends Configuration implements Tool, Runnable {


private String inputPath = null;
private String outputPath = null;

public IntSortJob(String inputPath,String outputPath){
this.inputPath = inputPath;
this.outputPath = outputPath;
}
public IntSortJob(){}

@Override
public Configuration getConf() {
// TODO Auto-generated method stub
return null;
}


@Override
public void setConf(Configuration arg0) {
// TODO Auto-generated method stub


}


@Override
public void run() {
try{
String[] args = {this.inputPath,this.outputPath};

start(args);

}catch (Exception e) {
e.printStackTrace();
}


}


private void start(String[] args)throws Exception{

ToolRunner.run(new UserJob(), args);
}


@Override
public int run(String[] args) throws Exception {
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(configuration);
fs.delete(new Path(args[1]),true);

Job job = new Job(configuration,"uniquejob");
job.setJarByClass(IntSortJob.class);

job.setMapperClass(IntSortMapper.class);
job.setReducerClass(IntSortReducer.class);

job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

boolean success = job.waitForCompletion(true);

return success?0:1;
}


}


package com.wy.hadoop.sort;


public class JobMain {


/**
* @param args
*/
public static void main(String[] args) {
if(args.length==2){
new Thread(new IntSortJob(args[0],args[1])).start();
}

}


}






7、本地创建3个文件，并把测试数据放入里面，然后将文件存放到hadoop hdfs中


8、打包程序到linux系统，执行hadoop jar 命令运行代码


9、查看结果