MapReduce框架中，输入类是如何定义和处理输入数据的？

在MapReduce编程模型中，输入类（InputFormat）是负责将输入数据切分为独立的单元（通常称为输入分片或InputSplit），并将这些分片转换为键值对（keyvalue pairs），供Mapper任务处理的关键组件，以下是关于MapReduce输入类的详细解析：

一、InputFormat概述

InputFormat是一个抽象类，它定义了如何从数据源读取数据并将其转换为适合MapReduce处理的键值对，MapReduce框架提供了多种内置的InputFormat实现，用于处理不同类型的输入数据，如文本文件、二进制文件、数据库表等，用户也可以根据需要自定义InputFormat。

二、InputFormat的核心方法

1、getSplits(JobContext job)：该方法负责将输入数据切分为多个InputSplit，每个InputSplit代表一个逻辑上的数据块，通常对应一个或多个物理文件的部分，切片的大小和数量取决于输入数据的大小、HDFS的块大小以及用户的配置。

2、createRecordReader(InputSplit split, TaskAttemptContext context)：该方法返回一个RecordReader实例，用于从给定的InputSplit中读取记录并生成键值对，RecordReader是实际执行数据读取操作的组件，它将InputSplit中的数据转换为适合Mapper处理的格式。

三、常见的InputFormat实现类

1、TextInputFormat：默认的文件输入格式，用于读取纯文本文件，它将每一行作为一个记录，键是行的字节偏移量（LongWritable类型），值是行的内容（Text类型）。

2、KeyValueTextInputFormat：类似于TextInputFormat，但它假设每一行由一个分隔符（通常是tab）分隔为两部分，第一部分为键，第二部分为值，如果没有分隔符，则整行作为键，值为空。

3、SequenceFileInputFormat：用于读取Hadoop的SequenceFile格式文件，SequenceFile是一种二进制文件格式，支持将任意类型的键值对序列化到文件中，SequenceFileInputFormat有两个子类：SequenceFileAsBinaryInputFormat（将键和值以BytesWritable类型读出）和SequenceFileAsTextInputFormat（将键和值以Text类型读出）。

4、NLineInputFormat：允许用户指定按多少行来切分InputSplit，这对于处理大文件中的特定行数非常有用。

5、CombineTextInputFormat：适用于小文件合并的场景，它可以将多个小文件合并成一个大的InputSplit，以减少MapTask的数量并提高处理效率。

6、自定义InputFormat：当内置的InputFormat无法满足需求时，用户可以自定义InputFormat，通过继承FileInputFormat类并重写getSplits和createRecordReader方法，用户可以定义自己的数据切分和读取逻辑。

四、自定义InputFormat示例

以下是一个自定义InputFormat的简单示例，该示例将整个文件的内容作为单个记录读取：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import java.io.IOException;
public class MyRecordReader extends RecordReader<NullWritable, BytesWritable> {
    private FileSplit fsplit;
    private Configuration conf;
    private boolean flag = false;
    private BytesWritable valueBtw = new BytesWritable();
    @Override
    public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        fsplit = (FileSplit) inputSplit;
        conf = taskAttemptContext.getConfiguration();
    }
    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if (!flag) {
            byte[] contentArr = new byte[(int) fsplit.getLength()];
            Path path = fsplit.getPath();
            FSDataInputStream in = FileSystem.get(conf).open(path);
            IOUtils.readFully(in, contentArr, 0, contentArr.length);
            in.close();
            valueBtw.set(contentArr, 0, contentArr.length);
            flag = true;
            return true;
        } else {
            return false;
        }
    }
    @Override
    public NullWritable getCurrentKey() throws IOException, InterruptedException {
        return NullWritable.get();
    }
    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        return valueBtw;
    }
    @Override
    public float getProgress() throws IOException, InterruptedException {
        return flag ? 1.0f : 0.0f;
    }
    @Override
    public void close() throws IOException {
        // No resources to release in this simple example
    }
}

在这个示例中，我们定义了一个MyRecordReader类，它继承了RecordReader<NullWritable, BytesWritable>类，这个RecordReader将从指定的FileSplit中读取整个文件的内容，并将其存储在一个BytesWritable对象中，它将返回一个键值对，其中键是NullWritable（表示没有键），值是包含文件内容的BytesWritable。

五、FAQs

Q1: MapReduce中的InputFormat是什么？它在MapReduce作业中扮演什么角色？

A1: MapReduce中的InputFormat是一个抽象类，它定义了如何将输入数据切分为独立的单元（InputSplit），并将这些单元转换为键值对（keyvalue pairs），供Mapper任务处理，InputFormat在MapReduce作业中扮演着至关重要的角色，因为它决定了数据的读取方式和切分策略，从而直接影响到MapReduce作业的性能和效率，不同的InputFormat实现可以处理不同类型的输入数据，如文本文件、二进制文件、数据库表等，用户还可以根据需要自定义InputFormat，以满足特定的数据处理需求。

Q2: 如何在MapReduce中自定义InputFormat？

A2: 在MapReduce中自定义InputFormat通常涉及以下几个步骤：你需要继承FileInputFormat类（或其子类），并重写getSplits方法和createRecordReader方法，getSplits方法负责将输入数据切分为多个InputSplit，而createRecordReader方法则负责从给定的InputSplit中读取记录并生成键值对，你需要定义自己的RecordReader类，该类负责实现具体的数据读取逻辑，在RecordReader中，你需要实现initialize方法（用于初始化）、nextKeyValue方法（用于读取下一条记录）、getCurrentKey方法（用于获取当前键）、getCurrentValue方法（用于获取当前值）以及getProgress方法（用于报告进度）和close方法（用于释放资源），你需要在MapReduce作业中设置使用自定义的InputFormat，这通常通过在Job配置中调用setInputFormatClass方法并传入自定义InputFormat类的class对象来实现，完成这些步骤后，你就可以在MapReduce作业中使用自定义的InputFormat来处理特定的输入数据了。