mapreduce的自定义OutputFormat代码

╰+攻爆jí腚メ 2022-12-14 03:36 187阅读 0赞

有时候，会用到将代码输出到特定的地方，所以需要自定义OutputFormat：

package application.mapreduce;
    
    import lombok.*;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FSDataOutputStream;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.RecordWriter;
    import org.apache.hadoop.mapreduce.TaskAttemptContext;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-12 20:13
     * @Version: 1.0
     * @Description:
     */
    public class FriendFileOutputFormat extends FileOutputFormat<Friend, Text> {
    
    
        @Override
        public RecordWriter<Friend, Text> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
            Configuration configuration = context.getConfiguration();
            FileSystem fileSystem = FileSystem.get(configuration);
    
            FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("D:/datas/result.dat"));
    
            return FriendRecordWriter.builder()
                    .fsDataOutputStream(fsDataOutputStream)
                    .build();
        }
    
        @EqualsAndHashCode(callSuper = true)
        @Data
        @AllArgsConstructor
        @NoArgsConstructor
        @Builder
        static class FriendRecordWriter extends RecordWriter<Friend, Text> {
    
            private FSDataOutputStream fsDataOutputStream;
    
            @Override
            public void write(Friend friend, Text text) throws IOException, InterruptedException {
                String[] fields = friend.getFriend().split("_");
                fsDataOutputStream.write(fields[0].getBytes());
                fsDataOutputStream.write(fields[1].getBytes());
                fsDataOutputStream.write((":" + text.toString() + "\n").getBytes());
            }
    
            @Override
            public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
                if (null != fsDataOutputStream){
                    fsDataOutputStream.close();
                }
            }
        }
    }