使用Hadoop的MapReduce实现数据排序

最近想系统学习大数据知识，在观看视频编写代码的时候，在数据排序的时候，出现了一些问题，一致于弄了好久才找到原因，现在记录下来，方便查看

数据输入格式：

按照我的代码逻辑，应该输出数据为

在代码处理时，计算结果却是

没有输出输入的数据，而是输出

最后保存在HDFS上的数据只是

1    1
2    2
3    3
4    4
5    5
6    6
7    7
8    8
9    9
10    10
11    11
12    12
13    13
14    14
15    15
16    16
17    17

我猜测是后面的数据覆盖了前面的写入的数据

我在网上查了一下，发现自己代码中使用了setCombinerClass（）

将这一行代码注释后，运行，真能跑出结果，然后就查setCombinerClass的用法：

别人也遇到过类似的问题：关于hadoop setCombinerClass 与 setReducerClass同时使用存在的问题。

是同时使用了setCombineClass()和setReducerClass()造成的

完整代码示例：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;


/**
 * FileName: SortedData
 * Author:   hadoop
 * Email:    [email protected]
 * Date:     18-10-6 上午10:54
 * Description:
 * 数字排序
 */

public class SortedData {
    /**
     * 使用Mapper将数据文件中的数据本身作为Mapper输出的key直接输出
     */

    public static class forSortedMapper extends Mapper<Object, Text, IntWritable, IntWritable> {
        private IntWritable mapperValue = new IntWritable(); //存放key的值
        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString(); //获取读取的值，转化为String
            mapperValue.set(Integer.parseInt(line)); //将String转化为Int类型
            context.write(mapperValue,new IntWritable(1)); //将每一条记录标记为（key，value） key--数字 value--出现的次数
                                                                //每出现一次就标记为（number，1）
        }
    }


/**
     * 使用Reducer将输入的key本身作为key直接输出
     */


 public static class forSortedReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{
        private IntWritable postion = new IntWritable(1); //存放名次
        @Override
        protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            for (IntWritable item :values){ //同一个数字可能出多次，就要多次并列排序
                context.write(postion,key); //写入名次和具体数字
                System.out.println(postion + "\t"+ key);
                postion = new IntWritable(postion.get()+1); //名次加1
            }
        }
    }


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


        Configuration conf = new Configuration(); //设置MapReduce的配置
        conf.set("mapred.job.tracker", "192.168.1.108:9000");
        String[] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs();
        if(otherArgs.length < 2){
            System.out.println("Usage: SortedData <in> [<in>...] <out>");
            System.exit(2);
        }

        //设置作业
        //Job job = new Job(conf);
        Job job = Job.getInstance(conf);
        job.setJarByClass(SortedData.class);
        job.setJobName("SortedData");
        //设置处理map,reduce的类
        job.setMapperClass(forSortedMapper.class);
        job.setReducerClass(forSortedReducer.class);
        //设置输入输出格式的处理
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        //设定输入输出路径
        for (int i = 0; i < otherArgs.length-1;++i){
            FileInputFormat.addInputPath(job,new Path(otherArgs[i]));
        }
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length-1]));
        System.exit(job.waitForCompletion(true)?0:1);
    }

}