http://www.rigongyizu.com/use-multiinputformat-read-different-files-in-one-job/
hadoop中提供了 MultiOutputFormat 能将结果数据输出到不同的目录，也提供了 FileInputFormat 来一次读取多个目录的数据，但是默认一个job只能使用 job.setInputFormatClass 设置使用一个inputfomat处理一种格式的数据。如果需要实现 在一个job中同时读取来自不同目录的不同格式文件 的功能，就需要自己实现一个 MultiInputFormat 来读取不同格式的文件了(原来已经提供了MultipleInputs)。
例如：有一个mapreduce job需要同时读取两种格式的数据，一种格式是普通的文本文件，用 LineRecordReader 一行一行读取；另外一种文件是伪XML文件，用自定义的AJoinRecordReader读取。
自己实现了一个简单的 MultiInputFormat 如下：
import org.apache.hadoop.io.LongWritable; |
import org.apache.hadoop.io.Text; |
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 org.apache.hadoop.mapreduce.lib.input.LineRecordReader; |
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; |
public class MultiInputFormat extends TextInputFormat { |
public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context) { |
RecordReader reader = null; |
String inputfile = ((FileSplit) split).getPath().toString(); |
String xmlpath = context.getConfiguration().get("xml_prefix"); |
String textpath = context.getConfiguration().get("text_prefix"); |
if (-1 != inputfile.indexOf(xmlpath)) { |
reader = new AJoinRecordReader(); |
} else if (-1 != inputfile.indexOf(textpath)) { |
reader = new LineRecordReader(); |
reader = new LineRecordReader(); |
} catch (IOException e) { |
其实原理很简单，就是在 createRecordReader 的时候，通过 ((FileSplit) split).getPath().toString() 获取到当前要处理的文件名，然后根据特征匹配，选取对应的 RecordReader 即可。xml_prefix和text_prefix可以在程序启动时通过 -D 传给Configuration。
比如某次执行打印的值如下：
这里只是通过简单的文件路径和标示符匹配来做，也可以采用更复杂的方法，比如文件名、文件后缀等。
接着在map类中，也同样可以根据不同的文件名特征进行不同的处理：
public void map(LongWritable offset, Text inValue, Context context) |
String inputfile = ((FileSplit) context.getInputSplit()).getPath() |
if (-1 != inputfile.indexOf(textpath)) { |
} else if (-1 != inputfile.indexOf(xmlpath)) { |
这种方式太土了，原来hadoop里面已经提供了 MultipleInputs 来实现对一个目录指定一个inputformat和对应的map处理类。
MultipleInputs.addInputPath(conf, new Path("/foo"), TextInputFormat.class, |
MultipleInputs.addInputPath(conf, new Path("/bar"), |
KeyValueTextInputFormat.class, MapClass2.class); |
posted on 2014-09-16 09:27
SIMONE 阅读(2730)
评论(0) 编辑 收藏 所属分类:
hadoop