如何在MapReduce中处理多个输入文件？
在MapReduce中处理多个输入文件的方法是使用MultipleInputs类。MultipleInputs类允许我们为每个输入文件指定不同的Mapper类，从而可以根据不同的输入文件执行不同的处理逻辑。
下面是一个使用MultipleInputs类处理多个输入文件的示例代码：
import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.MultipleInputs; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class MultipleInputsExample { public static void main(String[] args) throws Exception { // 创建一个新的MapReduce作业 Job job = Job.getInstance(); job.setJarByClass(MultipleInputsExample.class); job.setJobName("MultipleInputsExample"); // 设置多个输入文件路径和对应的Mapper类 MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, Mapper1.class); MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, Mapper2.class); // 设置Reducer类和输出键值对类型 job.setReducerClass(ReducerClass.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); // 设置输出文件路径 FileOutputFormat.setOutputPath(job, new Path(args[2])); // 提交作业并等待完成 System.exit(job.waitForCompletion(true) ? 0 : 1); } }
在上述代码中，我们首先创建了一个新的MapReduce作业，并设置了作业的名称和主类。然后，我们使用MultipleInputs类的addInputPath方法为每个输入文件指定路径和对应的Mapper类。在这个例子中，我们使用了两个输入文件，分别对应Mapper1类和Mapper2类。
接下来，我们设置了Reducer类和输出键值对的类型。在这个例子中，Reducer类为ReducerClass，输出键值对的类型为Text和LongWritable。
最后，我们设置了输出文件路径，并提交作业并等待完成。
下面是Mapper1类和Mapper2类的示例代码：
import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class Mapper1 extends Mapper<LongWritable, Text, Text, LongWritable> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 处理Mapper1的逻辑 // ... context.write(new Text("output_key"), new LongWritable(1)); } } public class Mapper2 extends Mapper<LongWritable, Text, Text, LongWritable> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 处理Mapper2的逻辑 // ... context.write(new Text("output_key"), new LongWritable(1)); } }
在这个示例中，Mapper1类和Mapper2类分别继承自Mapper类，并重写了map方法。在map方法中，我们可以根据具体的需求实现自己的逻辑。在这个例子中，我们简单地将每个输入记录映射为一个键值对（“output_key”, 1）。
可能的运行结果如下所示：
output_key 2
在这个例子中，我们使用了两个输入文件，并分别使用Mapper1类和Mapper2类处理。最终的输出结果是一个键值对（“output_key”, 2），表示"output_key"出现了两次。
通过使用MultipleInputs类，我们可以在MapReduce中处理多个输入文件，并根据不同的输入文件执行不同的处理逻辑。这样可以更灵活地处理不同来源的数据，并进行相应的处理和分析。
