hadoop编程小技巧（5）---自己定义输入文件格式类InputFormat

Hadoop代码測试环境：Hadoop2.4

应用：在对数据须要进行一定条件的过滤和简单处理的时候能够使用自己定义输入文件格式类。

Hadoop内置的输入文件格式类有：

1）FileInputFormat<K,V>这个是主要的父类，我们自己定义就直接使用它作为父类；

2）TextInputFormat<LongWritable,Text>这个是默认的数据格式类，我们一般编程，假设没有特别指定的话，一般都使用的是这个；key代表当前行数据距离文件開始的距离，value代码当前行字符串；

3）SequenceFileInputFormat<K,V>这个是序列文件输入格式，使用序列文件能够提高效率，可是不利于查看结果，建议在过程中使用序列文件，最后展示能够使用可视化输出；

4）KeyValueTextInputFormat<Text,Text>这个是读取以Tab（也即是 ）分隔的数据，每行数据假设以 分隔，那么使用这个读入，就能够自己主动把 前面的当做key，后面的当做value；

5）CombineFileInputFormat<K,V>合并大量小数据是使用；

6）MultipleInputs，多种输入，能够为每一个输入指定逻辑处理的Mapper；

原理：

InputFormat接口有两个重要的函数:

1)getInputSplits，用于确定输入分片，当我们继承FileInputFormat时，就能够忽略此函数，而使用FileInputFormat的此函数就可以；

2）createRecordReader ，针对数据怎样读取的类，定义输入文件格式，事实上也就是定义此类；

在每一个map函数中，最開始调用的都是nextKeyValue()函数，这个函数就是在RecordReader中定义的（我们自己定义RecordReader就是使用不同的实现而已），所以这里会调用我们指定的RecordReader中的nextKeyValue函数。这个函数就会处理或者说是初始化key和value，然后返回true，告知已经处理好了。接着就会调用getCurrentKey 和getCurrentValue获取当前的key和value值。最后，返回map，继续运行map逻辑。

自己定义输入文件格式类：

package fz.inputformat;

import java.io.IOException;

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.FileInputFormat;
/**
 * 自己定义输入文件读取类
 * 
 * @author fansy
 *
 */
public class CustomInputFormat extends FileInputFormat<Text, Text> {

	@Override
	public RecordReader<Text, Text> createRecordReader(InputSplit split,
			TaskAttemptContext context) throws IOException,
			InterruptedException {
		// TODO Auto-generated method stub
		return new CustomReader();
	}
}

这里看到假设继承了FileInputFormat 后，就不须要关心getInputSplits了，而仅仅须要定义RecordReader就可以。

自己定义RecordReader

package fz.inputformat;

//import java.io.BufferedReader;
import java.io.IOException;

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.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.util.LineReader;

public  class CustomReader extends RecordReader<Text ,Text>{
//	private BufferedReader in;
	private LineReader lr ;
	private Text key = new Text();
	private Text value = new Text();
	private long start ;
	private long end;
	private long currentPos;
	private Text line = new Text();
	@Override
	public void initialize(InputSplit inputSplit, TaskAttemptContext cxt)
			throws IOException, InterruptedException {
		FileSplit split =(FileSplit) inputSplit;
		Configuration conf = cxt.getConfiguration();
		Path path = split.getPath();
		FileSystem fs = path.getFileSystem(conf);
		FSDataInputStream is = fs.open(path);
		lr = new LineReader(is,conf);
		
		// 处理起始点和终止点
		start =split.getStart();
		end = start + split.getLength();
		is.seek(start);
		if(start!=0){
			start += lr.readLine(new Text(),0,
					(int)Math.min(Integer.MAX_VALUE, end-start));
		}
		currentPos = start;
	}

	// 针对每行数据进行处理
	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		if(currentPos > end){
			return false;
		}
		currentPos += lr.readLine(line);
		if(line.getLength()==0){
			return false;
		}
		if(line.toString().startsWith("ignore")){
			currentPos += lr.readLine(line);
		}
		
		String [] words = line.toString().split(",");
		// 异常处理
		if(words.length<2){
			System.err.println("line:"+line.toString()+".");
			return false;
		}
		key.set(words[0]);
		value.set(words[1]);
		return true;
		
	}

	@Override
	public Text getCurrentKey() throws IOException, InterruptedException {
		return key;
	}

	@Override
	public Text getCurrentValue() throws IOException, InterruptedException {
		return value;
	}

	@Override
	public float getProgress() throws IOException, InterruptedException {
		if (start == end) {
            return 0.0f;
        } else {
            return Math.min(1.0f, (currentPos - start) / (float) (end - start));
        }
	}

	@Override
	public void close() throws IOException {
		// TODO Auto-generated method stub
		lr.close();
	}
	
}

这里主要是两个函数，initial和nextKeyValue。

initial主要用于初始化，包含打开和读取文件，定义读取的进度等；

nextKeyValue则是针对每行数据（因为这里使用的是LineReader，所以每次读取的是一行，这里定义不同的读取方式，能够读取不同的内容），产生相应的key和value对，假设没有报错，则返回true。这里能够看到设置了一条规则，假设输入数据是以ignore開始的话就忽略，同一时候每行仅仅取得逗号前后的数据分别作为key和value。

实战：

输入数据：

ignore,2
a,3
ignore,4
c,1
c,2,3,2
4,3,2
ignore,34,2

定义主类，主类的Mapper是默认的Mapper，没有reducer。

package fz.inputformat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


public class FileInputFormatDriver extends Configured implements Tool{

	/**
	 * @param args
	 * @throws Exception 
	 */
	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		ToolRunner.run(new Configuration(), new FileInputFormatDriver(),args);
	}

	@Override
	public int run(String[] arg0) throws Exception {
		if(arg0.length!=2){
			System.err.println("Usage:
fz.inputformat.FileInputFormatDriver <in> <out>");
			return -1;
		}
		Configuration conf = getConf();
		
		Path in = new Path(arg0[0]);
		Path out= new Path(arg0[1]);
		out.getFileSystem(conf).delete(out, true);
		
		Job job = Job.getInstance(conf,"fileintputformat test job");
		job.setJarByClass(getClass());
		
		job.setInputFormatClass(CustomInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		job.setMapperClass(Mapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
//		job.setOutputKeyClass(LongWritable.class);
//		job.setOutputValueClass(VectorWritable.class);
		job.setNumReduceTasks(0);
//		System.out.println(job.getConfiguration().get("mapreduce.job.reduces"));
//		System.out.println(conf.get("mapreduce.job.reduces"));
		FileInputFormat.setInputPaths(job, in);
		FileOutputFormat.setOutputPath(job, out);
		
		return job.waitForCompletion(true)?0:-1;
	}

}

查看输出：

这里能够看到，ignore的数据已经被忽略掉了，同一时候每行仅仅输出了逗号前后的数据而已。

同一时候须要注意到：

这里有一行数据读入的是空字符串，这个临时还没找到原因。

总结：自己定义输入数据格式能够针对不同的数据做些过滤，进行一些简单的逻辑处理，有点相似map的功能，可是假设仅仅是这点功能的话，那全然能够使用map来代替了。事实上输入数据格式还有其它的功能，比方合并大量的小数据，以提高效率，这个在下篇再说。

分享，成长，快乐

转载请注明blog地址：http://blog.csdn.net/fansy1990