一个自动修正数据时间和补全缺失数据的MapReduce程序

原始数据如下图：

程序：

Mapper类：

public class DemoMapper extends Mapper<LongWritable,Text,IntWritable,Text>{
    IntWritable k = new IntWritable();
    Text v = new Text();
    
    @Override
    protected void map(LongWritable key,Text value,Mapper<LongWritable,Text,IntWritable,Text>.Context context)
            throws IOException,InterruptedException{
        String[] data = value.toString().split(",");
        k.set(Integer.parseInt(data[0]));
        try{
            v.set(Utils.getFixTime(data[1]));
            context.write(k,v);
        }catch(ParseException e){
            e.printStackTrace();
        }
            }
}

Reducer类：

public class DemoReducer extends Reducer<IntWritable,Text,NullWritable,Text>{
    Text v = new Text();
    
    @Override
    protected void reduce(IntWritable key,Iterable<Text> values,Reducer<IntWritable,Text,NullWritable,Text>.Context context)
            throws IOException,InterruptedException{
        TreeSet<Long> timeSet = new TreeSet<>();
        for(Text value : values){
            try{
                timeSet.add(getTime(value.toString()));
            }catch{
                e.printStackTrace();
            }
        }
        long tmp = -1;
        for(long time :timeSet){
            if(tmp == -1){
                v.set(key.toString()+","+getDate(time));
                context.write(NullWritable.get(),v);
            }else{
                if(time - tmp > 900000){
                    for（int i=0;i<= (time - tmp)/900000;i++）{
                        v.set(key.toString()+","+getDate(tmp+900000*i));
                    }
                }else{
                    v.set(key.toString()+","+getDate(time));
                    context.write(NullWritable.get(),v);
                }
            }
            tmp =time;
        }
            }
public static long getTime(String str)throws ParseException{
    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH-mm-ss");
    return simpleDateFormat.parse(str).getTime();
}

public static String getDate(long timetmp){
    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH-mm-ss");
    return simpleDateFormat.format(timeStamp);
}
}

Driver类：

public class DemoDriver{
    public static void main(String[] args)throws IllegalArgumentException,IOException,ClassNotFoundException,InterruptedException{
        if(args.length <2){
            System.err.println("you must input two argument!");
            System.exit(-1);
        }
        Configuration conf = Utils.getConf();
        Job job =Job.getInstance(conf, "fix time");
        job.setJarByClass(DemoDriver.class);
        job.setMapperClass(DemoMapper.class);
        job.setReducerClass(DemoReducer.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        job.setNumReduceTask(1);
        for(int i =0;i <args.length-1;i++){
            FileInputFormat.addInputPath(job,new Path(args[i]));
        }
        FileSystem.get(conf).delete(new Path(args[args.length-1]),true);
        FileOutputFormat.setOutputPath(job,new Path(args[args.length-1]));
        System.exit(job.waitForCompletion(true)?0:1);
    }
}