【转载】Hadoop自定义RecordReader

转自：http://www.linuxidc.com/Linux/2012-04/57831.htm

系统默认的LineRecordReader是按照每行的偏移量做为map输出时的key值，每行的内容作为map的value值，默认的分隔符是回车和换行。

现在要更改map对应的输入的<key,value>值，key对应的文件的路径（或者是文件名），value对应的是文件的内容（content）。

那么我们需要重写InputFormat和RecordReader，因为RecordReader是在InputFormat中调用的，当然重写RecordReader才是重点！

下面看代码InputFormat的重写：

1. public class chDicInputFormat extends FileInputFormat<Text,Text>
2. implements JobConfigurable{
3. private CompressionCodecFactory compressionCodecs = null;
4. public void configure(JobConf conf) {
5. compressionCodecs = new CompressionCodecFactory(conf);
6. }
7. /**
8. * @brief isSplitable 不对文件进行切分，必须对文件整体进行处理
9. *
10. * @param fs
11. * @param file
12. *
13. * @return false
14. */
15. protected boolean isSplitable(FileSystem fs, Path file) {
16. //  CompressionCodec codec = compressionCodecs.getCode(file);
17. return false;//以文件为单位，每个单位作为一个split，即使单个文件的大小超过了64M，也就是Hadoop一个块得大小，也不进行分片
18. }
19. public RecordReader<Text,Text> getRecordReader(InputSplit genericSplit,
20. JobConf job, Reporter reporter) throws IOException{
21. reporter.setStatus(genericSplit.toString());
22. return new chDicRecordReader(job,(FileSplit)genericSplit);
23. }
24. }

下面来看RecordReader的重写：

1. public class chDicRecordReader implements RecordReader<Text,Text> {
2. private static final Log LOG = LogFactory.getLog(chDicRecordReader.class.getName());
3. private CompressionCodecFactory compressionCodecs = null;
4. private long start;
5. private long pos;
6. private long end;
7. private byte[] buffer;
8. private String keyName;
9. private FSDataInputStream fileIn;
10. public chDicRecordReader(Configuration job,FileSplit split) throws IOException{
11. start = split.getStart(); //从中可以看出每个文件是作为一个split的
12. end = split.getLength() + start;
13. final Path path = split.getPath();
14. keyName = path.toString();
15. LOG.info("filename in hdfs is : " + keyName);
16. final FileSystem fs = path.getFileSystem(job);
17. fileIn = fs.open(path);
18. fileIn.seek(start);
19. buffer = new byte[(int)(end - start)];
20. this.pos = start;
21. }
22. public Text createKey() {
23. return new Text();
24. }
25. public Text createValue() {
26. return new Text();
27. }
28. public long getPos() throws IOException{
29. return pos;
30. }
31. public float getProgress() {
32. if (start == end) {
33. return 0.0f;
34. } else {
35. return Math.min(1.0f, (pos - start) / (float)(end - start));
36. }
37. }
38. public boolean next(Text key, Text value) throws IOException{
39. while(pos < end) {
40. key.set(keyName);
41. value.clear();
42. fileIn.readFully(pos,buffer);
43. value.set(buffer);
44. //      LOG.info("---内容: " + value.toString());
45. pos += buffer.length;
46. LOG.info("end is : " + end  + " pos is : " + pos);
47. return true;
48. }
49. return false;
50. }
51. public void close() throws IOException{
52. if(fileIn != null) {
53. fileIn.close();
54. }
55. }
56. }

通过上面的代码，然后再在main函数中设置InputFormat对应的类，就可以使用这种新的读入格式了。