（1）以怎样的方式从分片中读取一条记录，每读取一条记录都会调用RecordReader类；
（2）系统默认的RecordReader是LineRecordReader，如TextInputFormat；而SequenceFileInputFormat的RecordReader是SequenceFileRecordReader；
（3）LineRecordReader是用每行的偏移量作为map的key，每行的内容作为map的value；
（4）应用场景：自定义读取每一条记录的方式；自定义读入key的类型，如希望读取的key是文件的路径或名字而不是该行在文件中的偏移量。
自定义RecordReader：
（1）继承抽象类RecordReader，实现RecordReader的一个实例；
（2）实现自定义InputFormat类，重写InputFormat中createRecordReader（）方法，返回值是自定义的RecordReader实例；
（3）配置job.setInputFormatClass()设置自定义的InputFormat实例；
源码见org.apache.mapreduce.lib.input.TextInputFormat类；
RecordReader例子：
应用场景：
数据：
1
2
3
4
5
6
7
......
要求：分别计算奇数行与偶数行数据之和
奇数行综合：10+30+50+70=160
偶数行综合：20+40+60=120
新建项目TestRecordReader，包com.recordreader,
源代码MyMapper.java：
package com.recordreader;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class MyMapper extends Mapper {
@Override
protected void map(LongWritable key, Text value,Context context)
throws IOException, InterruptedException {
// TODO Auto-generated method stub
context.write(key, value);
}
}
源代码MyPartitioner.java：
package com.recordreader;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
public class MyPartitioner extends Partitioner {
@Override
public int getPartition(LongWritable key, Text value, int numPartitions) {
// TODO Auto-generated method stub
if(key.get() % 2 == 0){
key.set(1);
return 1;
}
else {
key.set(0);
return 0;
}
}
}
源代码MyReducer.java：
package com.recordreader;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class MyReducer extends Reducer {
@Override
protected void reduce(LongWritable key, Iterable value,Context context)
throws IOException, InterruptedException {
// TODO Auto-generated method stub
int sum = 0;
for(Text val: value){
sum += Integer.parseInt(val.toString());
}
Text write_key = new Text();
IntWritable write_value = new IntWritable();
if(key.get() == 0)
write_key.set("odd:");
else
write_key.set("even:");
write_value.set(sum);
context.write(write_key, write_value);
}
}
源代码MyRecordReader.java：
package com.recordreader;
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.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.util.LineReader;
public class MyRecordReader extends RecordReader {
private long start;
private long end;
private long pos;
private FSDataInputStream fin = null;
private LongWritable key = null;
private Text value = null;
private LineReader reader = null;
@Override
public void close() throws IOException {
// TODO Auto-generated method stub
fin.close();
}
@Override
public LongWritable getCurrentKey() throws IOException,
InterruptedException {
// TODO Auto-generated method stub
return key;
}
@Override
public Text getCurrentValue() throws IOException, InterruptedException {
// TODO Auto-generated method stub
return value;
}
@Override
public float getProgress() throws IOException, InterruptedException {
// TODO Auto-generated method stub
return 0;
}
@Override
public void initialize(InputSplit inputSplit, TaskAttemptContext context)
throws IOException, InterruptedException {
// TODO Auto-generated method stub
FileSplit fileSplit = (FileSplit)inputSplit;
start = fileSplit.getStart();
end = start + fileSplit.getLength();
Configuration conf = context.getConfiguration();
Path path = fileSplit.getPath();
FileSystem fs = path.getFileSystem(conf);
fin = fs.open(path);
fin.seek(start);
reader = new LineReader(fin);
pos = 1;
}
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
// TODO Auto-generated method stub
if(key == null)
key = new LongWritable();
key.set(pos);
if(value == null)
value = new Text();
if(reader.readLine(value) == 0)
return false;
pos++;
return true;
}
}
源代码MyFileInputFormat.java：
package com.recordreader;
import java.io.IOException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
public class MyFileInputFormat extends FileInputFormat {
@Override
public RecordReader createRecordReader(InputSplit arg0,
TaskAttemptContext arg1) throws IOException, InterruptedException {
// TODO Auto-generated method stub
return new MyRecordReader();
}
@Override
protected boolean isSplitable(JobContext context, Path filename) {
// TODO Auto-generated method stub
return false;
}
}
源代码TestRecordReader.java：
package com.recordreader;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class TestRecordReader {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException{
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage: wordcount ");
System.exit(2);
}
Job job = new Job(conf, "word count");
job.setJarByClass(TestRecordReader.class);
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
job.setPartitionerClass(MyPartitioner.class);
job.setNumReduceTasks(2);
job.setInputFormatClass(MyFileInputFormat.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
