寫出MapReduce程序完成以下功能.

input1：

2012-3-1 a
2012-3-2 b
2012-3-3 c
2012-3-4 d
2012-3-5 a
2012-3-6 b
2012-3-7 c
2012-3-3 c

input2：

2012-3-1 b
2012-3-2 a
2012-3-3 b
2012-3-4 d
2012-3-5 a
2012-3-6 c
2012-3-7 d
2012-3-3 c

目標操作實現結果：

2012年3月3日 c
2012年3月7日 c
2012年3月6日 b
2012年3月5日 a
2012年3月4日 d
2012年3月3日 c
2012年3月2日 b
2012年3月1日 a
2012年3月3日 c
2012年3月7日 d
2012年3月6日 c
2012年3月5日 a
2012年3月4日 d
2012年3月3日 b
2012年3月2日 a
2012年3月1日 b
 

代碼如下（由於水平有限，不保證完全正確，如果發現錯誤歡迎指正）：

package one;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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;

 
public class TestYear {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration config = new  Configuration();
        config.set("fs.defaultFS", "hdfs://192.168.0.100:9000");
        config.set("yarn.resourcemanager.hostname", "192.168.0.100
 
");
        
        FileSystem fs = FileSystem.get(config);
        
        Job job = Job.getInstance(config);
        
        job.setJarByClass(TestYear.class);
        
        //設置所用到的map類
        job.setMapperClass(myMapper.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(Text.class);
        
        //設置用到的reducer類
        job.setReducerClass(myReducer.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        
        //設置輸出地址
        FileInputFormat.addInputPath(job, new Path("/zhoukao3/"));
        
        Path path = new Path("/output1/");
        
        if(fs.exists(path)){
            fs.delete(path, true);
        }
        
        //指定文件的輸出地址
        FileOutputFormat.setOutputPath(job, path);
        
        //啟動處理任務job
        boolean completion = job.waitForCompletion(true);
        if(completion){
            System.out.println("Job Success!");
        }
    }
    
    public static class myMapper extends Mapper<LongWritable, Text, NullWritable , Text>{

        @Override
        protected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {
            String values=value.toString();
            String words[]=values.split("[-]| ");//2012,3,1,a
            String s=words[0]+"年"+words[1]+"月"+words[2]+"日"+"  "+words[3];
            context.write(NullWritable.get(),new Text(s));
        }
    }
    
    public static class myReducer extends Reducer<NullWritable , Text,NullWritable , Text>{

        @Override
        protected void reduce(NullWritable key, Iterable<Text> values,Context context)throws IOException, InterruptedException {
            for (Text value  : values) {
                context.write(key, value);
            }
        }
        
    }
}

小結：把value-list作為map的value輸出，這樣就不會排序和去重，然後reduce再去接收並且context.write()出來，需要註意的是-號是特殊字符，需要做分割處理，所以可以加上\\或者[ ]，註意點這些小細節就能完成最終的效果。

寫出MapReduce程序完成以下功能