如何自定义 MapReduce 的 InputFormat 和 OutputFormat？

推荐答案

自定义 InputFormat

1. 继承 FileInputFormat 类：

   public class CustomInputFormat extends FileInputFormat<LongWritable, Text> {
       @Override
       public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context) {
           return new CustomRecordReader();
       }
   }

2. 实现 RecordReader 类：

   // www.javascriptcn.com code example
   public class CustomRecordReader extends RecordReader<LongWritable, Text> {
       private LongWritable key;
       private Text value;
       private LineReader reader;
   
       @Override
       public void initialize(InputSplit split, TaskAttemptContext context) throws IOException {
           // 初始化读取器
           FileSplit fileSplit = (FileSplit) split;
           Configuration conf = context.getConfiguration();
           Path file = fileSplit.getPath();
           FileSystem fs = file.getFileSystem(conf);
           FSDataInputStream in = fs.open(file);
           reader = new LineReader(in, conf);
       }
   
       @Override
       public boolean nextKeyValue() throws IOException {
           // 读取下一行数据
           key = new LongWritable();
           value = new Text();
           int bytesRead = reader.readLine(value);
           if (bytesRead == 0) {
               return false;
           }
           key.set(reader.getPos());
           return true;
       }
   
       @Override
       public LongWritable getCurrentKey() {
           return key;
       }
   
       @Override
       public Text getCurrentValue() {
           return value;
       }
   
       @Override
       public float getProgress() throws IOException {
           // 返回读取进度
           return 0;
       }
   
       @Override
       public void close() throws IOException {
           if (reader != null) {
               reader.close();
           }
       }
   }

自定义 OutputFormat

1. 继承 FileOutputFormat 类：

   // www.javascriptcn.com code example
   public class CustomOutputFormat extends FileOutputFormat<Text, IntWritable> {
       @Override
       public RecordWriter<Text, IntWritable> getRecordWriter(TaskAttemptContext context) throws IOException {
           // 获取输出路径
           Path outputDir = FileOutputFormat.getOutputPath(context);
           Path file = new Path(outputDir, "part-r-" + context.getTaskAttemptID().getId());
           FileSystem fs = file.getFileSystem(context.getConfiguration());
           FSDataOutputStream out = fs.create(file);
   
           return new CustomRecordWriter(out);
       }
   }

2. 实现 RecordWriter 类：

   // www.javascriptcn.com code example
   public class CustomRecordWriter extends RecordWriter<Text, IntWritable> {
       private FSDataOutputStream out;
   
       public CustomRecordWriter(FSDataOutputStream out) {
           this.out = out;
       }
   
       @Override
       public void write(Text key, IntWritable value) throws IOException {
           // 自定义写入逻辑
           out.writeBytes(key.toString() + "\t" + value.get() + "\n");
       }
   
       @Override
       public void close(TaskAttemptContext context) throws IOException {
           out.close();
       }
   }

本题详细解读

自定义 InputFormat

在 Hadoop 中，InputFormat 负责定义如何读取输入数据并将其分割成 InputSplit，然后由 RecordReader 将每个 InputSplit 转换为键值对。自定义 InputFormat 通常需要以下步骤：

1. 继承 FileInputFormat：FileInputFormat 是 Hadoop 提供的抽象类，通常用于处理文件输入。自定义的 InputFormat 需要继承此类，并实现 createRecordReader 方法。

2. 实现 RecordReader：RecordReader 负责从 InputSplit 中读取数据并生成键值对。自定义的 RecordReader 需要实现 initialize、nextKeyValue、getCurrentKey、getCurrentValue、getProgress 和 close 方法。

自定义 OutputFormat

OutputFormat 负责定义如何将输出数据写入存储系统。自定义 OutputFormat 通常需要以下步骤：

1. 继承 FileOutputFormat：FileOutputFormat 是 Hadoop 提供的抽象类，通常用于处理文件输出。自定义的 OutputFormat 需要继承此类，并实现 getRecordWriter 方法。

2. 实现 RecordWriter：RecordWriter 负责将键值对写入输出文件。自定义的 RecordWriter 需要实现 write 和 close 方法。

通过自定义 InputFormat 和 OutputFormat，可以灵活地处理各种输入输出格式，满足特定的业务需求。