MapReduce之自定义outputFormat

public class Custom_OutputFormat extends FileOutputFormat<Text, NullWritable> {



    @Override
    public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {


        Configuration configuration = context.getConfiguration();

        FileSystem fileSystem = FileSystem.get(configuration);

        //两个输出路径
        FSDataOutputStream fsDataOutputStream1 = fileSystem.create(new Path("E:\\2019大数据课程\\DeBug\\测试结果\\outputformat1\\1.txt"));

        FSDataOutputStream fsDataOutputStream2 = fileSystem.create(new Path("E:\\2019大数据课程\\DeBug\\测试结果\\outputformat1\\2.txt"));

        Custom_RecordWriter custom_recordWriter = new Custom_RecordWriter(fsDataOutputStream1, fsDataOutputStream2);




        return custom_recordWriter;

    }
}

public class Custom_RecordWriter extends RecordWriter<Text, NullWritable> {


    FSDataOutputStream out1 = null;

    FSDataOutputStream out2 = null;


    @Override
    public String toString() {
        return "Custom_RecordWriter{" +
                "out1=" + out1 +
                ", out2=" + out2 +
                '}';
    }

    public FSDataOutputStream getOut1() {
        return out1;
    }

    public void setOut1(FSDataOutputStream out1) {
        this.out1 = out1;
    }

    public FSDataOutputStream getOut2() {
        return out2;
    }

    public void setOut2(FSDataOutputStream out2) {
        this.out2 = out2;
    }

    public Custom_RecordWriter() {
    }

    public Custom_RecordWriter(FSDataOutputStream out1, FSDataOutputStream out2) {
        this.out1 = out1;
        this.out2 = out2;
    }



    /**
     * 写入数据的方法
     * @param key   要写入的key值
     * @param value   要写入的value
     * @throws IOException
     * @throws InterruptedException
     */

    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {

               if (key.toString().split("\t")[9].equals("0")){
                   //好评
                   out1.write(key.toString().getBytes());


               }else{


                   //中评和差评
                   out2.write(key.toString().getBytes());
                   out2.write("\r\n".getBytes());

               }



    }

    //  关闭
    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {

        if (out1!=null){
            out1.close();
        }

        if (out2!=null){
            out2.close();
        }



    }
}

public class Custom_Mapper extends Mapper<LongWritable, Text,Text, NullWritable> {



    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {


        context.write(value,NullWritable.get());


    }

public class Custom_Driver {


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


        Job job = Job.getInstance(new Configuration(), "Customer_Driver");

        job.setJarByClass(Custom_Driver.class);

        //设置输入
        job.setInputFormatClass(TextInputFormat.class);

        TextInputFormat.addInputPath(job,new Path("E:\\2019大数据课程\\DeBug\\测试\\ordercomment.csv"));

        job.setMapperClass(Custom_Mapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setOutputFormatClass(Custom_OutputFormat.class);

        // 这里path的路径可以任意设置,因为我们在自定义outPutFormat中已经将输出路径确定
        Custom_OutputFormat.setOutputPath(job,new Path("E:\\2019大数据课程\\DeBug\\测试结果\\outputformat1"));

        boolean b = job.waitForCompletion(true);

        System.out.println(b?0:1);


    }
}