/**
 * 需求：不同结果输出到不同的文件夹下
 * 统计平均成绩   成绩低于60分的写到/chengji/bujige/bujige.txt    成绩不低于60分的写到/chengji/jige/jige.txt
 */
public class DiffDic {

    static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

        Text mapKey = new Text();
        IntWritable mapValue = new IntWritable();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] datas = value.toString().split(",");
            String k = datas[0] + "\t" + datas[1];
            mapKey.set(k);

            for (int i=2; i<datas.length; i++){
                mapValue.set(Integer.parseInt(datas[i]));
                context.write(mapKey, mapValue);
            }
        }
    }

    static class MyReducer extends Reducer<Text, IntWritable, Text, DoubleWritable>{

        DoubleWritable value = new DoubleWritable();

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            int count = 0;
            for (IntWritable v:values) {
                sum+=v.get();
                count++;
            }
            double avg = (double)sum/count;
            value.set(avg);
            context.write(key,value);
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        args = new String[]{"/scorein","/scoreout"};

        Configuration conf = new Configuration();

        //1 获取 Job 对象
        Job job = Job.getInstance(conf);

        //2 设置 jar 存储位置
        job.setJarByClass(DiffDic.class);

        //3 关联 Map 和 Reduce 类
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);

        //4 设置 Mapper 阶段输出数据的 key 和 value 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //5 设置最终数据输出的 key 和 value 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        // 指定自定义输出
        job.setOutputFormatClass(DiffDicOutputFormat.class);

        //6 设置输入路径和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //7 提交 job
        //job.submit();//此方法不会打印日志
        boolean result = job.waitForCompletion(true);

        System.exit(result?0:1);
    }
}

public class DiffDicOutputFormat extends FileOutputFormat<Text, DoubleWritable> {

    //获取 RecordWriter 对象
    @Override
    public RecordWriter<Text, DoubleWritable> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
        FileSystem fs = FileSystem.get(job.getConfiguration());

        return new DiffRecordWriter(fs);
    }
}

public class DiffRecordWriter extends RecordWriter<Text, DoubleWritable> {
    FSDataOutputStream out01 = null;
    FSDataOutputStream out02 = null;
    FileSystem fs;


    public DiffRecordWriter(FileSystem fs) throws IOException {
        this.fs = fs;
        out01 = fs.create(new Path("/kaoshi/jige/jige.txt"));
        out02 = fs.create(new Path("/kaoshi/bujige/bujige.txt"));

    }

    // 真正向外写出的方法

    /**
     * 需求：不同成绩输出到不同文件夹。--则需要两个输出流
     * 流可通过 filesystem 对象，但 filesystem 对象需要通过配置文件获取
     * @param key
     * @param value
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void write(Text key, DoubleWritable value) throws IOException, InterruptedException {
        double score = value.get();
        if (score >= 60){
            out01.write((key.toString() + "\t" + Double.toString(score) + "\n").getBytes("utf-8"));
        }else {
            out02.write((key.toString() + "\t" + Double.toString(score) + "\n").getBytes("utf-8"));
        }
    }

    //关闭资源的方法
    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        out01.close();
        out02.close();
    }
}