首先说一下Hadoop中预定义的Mapper 与Reducer
InputFormat接口决定了输入文件如何被hadoop分块(split up)与接受。
TextInputFormat是InputFormat的默认实现，对于输入数据中没有明确的key值时非常有效，TextInputFormat返回的key值为字符在输入块中的行数，value为这行的内容。
其他InputFormat的子类还有
KeyValueTextInputFormat（键：Text，值:Text）他的分割符默认为tab("\t")，我们可以通过key.value.separator.input.line.property设置
SequenceFileInputFormat<K,V>（键和值都是由用户定义）
NLineInputFormat（键：LongWritable，值：Text）等
你的MapReduce程序如果要是用KeyValueTextInputFormat作为输入格式，我们可以这样做：
conf.setInputFormat(KeyValueTextInputFormat.class);
package org.apache.hadoop.mapred;
import java.io.IOException;
import org.apache.hadoop.fs.FileSystem;
public interface InputFormat<K, V> {
InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;
RecordReader<K, V> getRecordReader(InputSplit split,
JobConf job,
Reporter reporter) throws IOException;
}
以上是InputFormat的定义，里边有两个方法。
第一个方法的功能是确认所有的文件为输入数据并且把他们分成块(splits)。每一个任务都被分配一个块(split)。
第二个方式的功能是提供一个对象用来遍历给定块（split）中的记录，并且把记录解析成先前定义的key和value类型。
一般来说getSplits()方法不用我们去管。其实以上所举出的InputFormat的子类都是继承子FileInputFormat的，FileInputFormat实现了getSplits方法，把getRecordReader()方法留给子类去实现。
我们在使用FileInputFormat时，我们主要的精力在定制合适的RecordReader类上，因为他负责如何将splits解析为records，将records解析成为key/value对。
package org.apache.hadoop.mapred;
import java.io.IOException;
import java.io.DataInput;
public interface RecordReader<K, V> {
boolean next(K key, V value) throws IOException;
K createKey();
V createValue();
long getPos() throws IOException;
public void close() throws IOException;
float getProgress() throws IOException;
}
以上为RecordReader的签名，我们一般从现有的RecordReader的子类中来定制自己的RecordReader。主要是实现next()方法。
比如LineRecordReader继承了RecordReader<LongWritable, Text>,TextInputFormat使用了这个Reader。KeyValueLineRecordReader用在了KeyValueTextInputFormat类中。
下面我们自己实现一个RecordReader类，她将要解析的key为Text，value为URLWritable（自己定义）
首先我们需要定义URLWritable这个类，有了这个以后我们就可以定义我们的Reader了
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueLineRecordReader;
import org.apache.hadoop.mapred.RecordReader;
public class TimeUrlLineRecordReader implements RecordReader<Text, URLWritable> {
private KeyValueLineRecordReader lineReader;
private Text lineKey, lineValue;
public TimeUrlLineRecordReader(JobConf job, FileSplit split) throws IOException {
lineReader = new KeyValueLineRecordReader(job, split);
lineKey = lineReader.createKey();
lineValue = lineReader.createValue();
}
@Override
public boolean next(Text key, URLWritable value) throws IOException {
// TODO Auto-generated method stub
if (!lineReader.next(lineKey, lineValue)) {
return false;
}
key.set(lineKey);
value.set(lineValue.toString());
return true;
}
@Override
public Text createKey() {
// TODO Auto-generated method stub
return new Text("");
}
@Override
public URLWritable createValue() {
// TODO Auto-generated method stub
return new URLWritable();
}
@Override
public long getPos() throws IOException {
// TODO Auto-generated method stub
return lineReader.getPos();
}
@Override
public void close() throws IOException {
// TODO Auto-generated method stub
lineReader.close();
}
@Override
public float getProgress() throws IOException {
// TODO Auto-generated method stub
return lineReader.getProgress();
}
}
class URLWritable implements Writable {
protected URL url;
public URLWritable() {}
public URLWritable(URL url) {
this.url = url;
}
@Override
public void write(DataOutput out) throws IOException {
// TODO Auto-generated method stub
out.writeUTF(url.toString());
}
@Override
public void readFields(DataInput in) throws IOException {
// TODO Auto-generated method stub
url = new URL(in.readUTF());
}
public void set(String s) throws MalformedURLException {
url = new URL(s);
}
}
可以看到，我们在TimeUrlLineRecordReader类中创建了一个KeyValueLineRecordReader对象，然后在实现getPos()、getProgress()、close()方法是直接是哦那个他对应的方法即可。
在next()方法中，我Text类型的lineValue转型为URLWritable类型了。
这时我们就可呀使用这个RecordReader了
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
public class TimeUrlTextInputFormat extends FileInputFormat<Text, URLWritable> {
@Override
public RecordReader<Text, URLWritable> getRecordReader(InputSplit split,
JobConf job, Reporter reporter) throws IOException {
// TODO Auto-generated method stub
return new TimeUrlLineRecordReader(job, (FileSplit)split);
}
}
OutputFormat和InputFormat类似，但是有些地方不一样。
和InputFormat一样，所有的OutputFormat大多数继承自FileOutputFormat，但是NullOutputForma和DBOutputFormat。他们是为专门领域的应用程序预留的。
TextOutputFormat是默认的输入格式，key value 用一个tab分开，分割符也可以通过mapred.textoutputformat.separator属性设置
TextOutputFormat的输出格式可以被KeyValueTextInputFormat接受
如果输出的key类型为NullWritable的输出格式可以被TextInputFormat接受。在这中情况下key没有被输出来，也没有分割符。
如果不想让reduce程序有任何输出我们可以把输出格式设置为NullOutputFormat。阻止hadoop的输出在reducer用自己的方式输出文件而不许要hadoop框架些任何额外文件是十分有用。
SequenceFileOutuputFormat它把输出写入到一个串文件中（sequence files），这样我么就可以用SequenceFileInputFormat读会数据。这在处理多个联系mapreduce任务时十分有效。
PS:hadoop中预定义的InputFormat与OutputFormat