No Reply , Posted in Hadoop on December 2, 2012

在Hadoop中，InputFormat类用来生成可供Mapper处理的<key, value>键值对。当数据传送给Mapper时，Mapper会将输入分片传送到InputFormat上，InputFormat调用getRecordReader()方法生成RecordReader，RecordReader再创建可供map函数处理的键值对<K1, V1>。

Hadoop预定义了多种方法将不同类型的输入数据转化为map能够处理的键值对。比如，TextInputFormat，Hadoop中默认的输入方法，会将每行数据生成一条记录，其中key值为每条记录在分片中的字节偏移量，value则为每行的内容。

在Hadoop预定义的InputFormat中，并没有处理CSV文件的方法。CSV文件的本质其实是用逗号分隔开的文本文件。一种很直观的处理方法是：将CSV文件作为文本文件处理，使用TextInputFormat将文件按行传入map函数，在map函数中再按照CSV文件的格式进行处理。但这样很容易将数据格式的处理逻辑与业务处理逻辑混淆在一起，并且出现很多copy-and-pasted的代码。

实际上，可以写一个自己的InputFormat以及RecordReader类，专门用来处理CSV文件的输入，直接传递给map函数解析后的数据。

1 数据结构

我们传递给map函数一个ArrayWritable（A Writable for arrays containing instances of a class），元素类型为Text，即CSV文件每一行各个字段的数据。数据结构如下：

代码1：TextArrayWritable.java

1. public class TextArrayWritable extends ArrayWritable {
2. public TextArrayWritable() {
3. super(Text.class);
4. }
6. public TextArrayWritable(Text[] strings) {
7. super(Text.class, strings);
8. }
9. }

2 CSVInputFormat

FileInputFormat是所有使用文件作为其数据源的InputFormat实现的基类。它提供了两个功能：一是定义哪些文件包含在一个作业的输入中，另一个是为输入文件生成分片（Input Splits）。而把分片分割成记录的事情交由其子类来完成。所以CSVInputFormat类的实现上，同样是继承InputFormat类，并只需要简单的重写createRecordReader和isSplitable即可。

代码2：CSVInputFormat.java

1. public class CSVInputFormat
2. extends FileInputFormat<LongWritable, TextArrayWritable>{
3. public static final String CSV_TOKEN_SEPARATOR_CONFIG
4. = "csvinputformat.token.delimiter";
6. @Override
7. protected boolean isSplitable(JobContext context, Path filename) {
8. CompressionCodec codec =
9. new CompressionCodecFactory(context.getConfiguration())
10. .getCodec(filename);
11. return codec == null;
12. }
14. @Override
15. public RecordReader<LongWritable, TextArrayWritable> createRecordReader(
16. InputSplit split, TaskAttemptContext context)
17. throws IOException, InterruptedException {
18. String csvDelimiter = context.getConfiguration()
19. .get(CSV_TOKEN_SEPARATOR_CONFIG);
20. Character separator = null;
21. if (csvDelimiter != null && csvDelimiter.length() == 1) {
22. separator = csvDelimiter.charAt(0);
23. }
24. return new CSVRecordReader(separator);
25. }
26. }

其中csvinputformat.token.delimiter是可在配置文件中配置的CSV输入文件分隔符，createRecordReader完成的工作只是从配置文件中得到分隔符，调用真正对CSV文件分片进行处理，并生成键值对的CSVRecordReader函数，并返回RecordReader对象。

3CSVRecordReader

对于CSVRecordReader，要实现的功能无非就是将CSV文件中每一行的各字段提取出来，并将各字段作为TextArrayWritable类型的数据结构传递给map函数。

在Hadoop中有一个LineRecordReader类，它将文本文件每一行的内容作为值返回，类型为Text。所以可以直接在CSVRecordReader中使用LineRecordReader，将LineRecordReader返回的每一行再次进行处理。在CSV文件的处理上，这里用到了OpenCSV对CSV文件的每一行进行解析，具体可参见这里。

下面是CSVRecordReader的实现代码。除了CSV文件的解析、nextKeyValue()方法和getCurrentValue()方法外，大部分方法都直接调用LineRecordReader实例的相应方法。毕竟我们是踩在巨人的肩膀上继续前进嘛。O(∩_∩)O~

代码3：CSVRecordReader.java

1. public class CSVRecordReader
2. extends RecordReader<LongWritable, TextArrayWritable> {
4. private LineRecordReader lineReader;
5. private TextArrayWritable value;
6. private CSVParser parser;
8. // 新建CSVParser实例，用来解析每一行CSV文件的每一行
9. public CSVRecordReader(Character delimiter) {
10. this.lineReader = new LineRecordReader();
11. if (delimiter == null) {
12. this.parser = new CSVParser();
13. }
14. else {
15. this.parser = new CSVParser(delimiter);
16. }
17. }
19. // 调用LineRecordReader的初始化方法，寻找分片的开始位置
20. @Override
21. public void initialize(InputSplit split, TaskAttemptContext context)
22. throws IOException, InterruptedException {
23. lineReader.initialize(split, context);
24. }
26. // 使用LineRecordReader来得到下一条记录（即下一行）。
27. // 如果到了分片（Input Split）的尾部，nextKeyValue将返回NULL
28. @Override
29. public boolean nextKeyValue()
30. throws IOException, InterruptedException {
31. if (lineReader.nextKeyValue()) {
32. //如果有新记录，则进行处理
33. loadCSV();
34. return true;
35. }
36. else {
37. value = null;
38. return false;
39. }
40. }
42. @Override
43. public LongWritable getCurrentKey() throws IOException,
44. InterruptedException {
45. return lineReader.getCurrentKey();
46. }
48. @Override
49. public TextArrayWritable getCurrentValue() throws IOException,
50. InterruptedException {
51. return value;
52. }
54. @Override
55. public float getProgress() throws IOException, InterruptedException {
56. return lineReader.getProgress();
57. }
59. @Override
60. public void close() throws IOException {
61. lineReader.close();
62. }
64. // 对CSV文件的每一行进行处理
65. private void loadCSV() throws IOException {
66. String line = lineReader.getCurrentValue().toString();
67. // 通过OpenCSV将解析每一行的各字段
68. String[] tokens = parser.parseLine(line);
69. value = new TextArrayWritable(convert(tokens));
70. }
72. // 将字符串数组批量处理为Text数组
73. private Text[] convert(String[] tokens) {
74. Text[] t = new Text[tokens.length];
75. for (int i = 0; i < t.length; i++) {
76. t[i] = new Text(tokens[i]);
77. }
78. return t;
79. }
80. }

4简单的应用

用于处理CSV文件输入的InputFormat已经写完了，现在构造一个简单的应用场景，来试验下这个CSVInputFormat。

假设有这样一些数据，每一列第一个字段为一个标识，后面为随机产生的数字，标识各不相同，求每一行标识后的数字之和并输出，输出格式为：每一行为标识和数字和。

由于标识没有重复，并且逻辑比较简单，这里只写一个Mapper即可，不需要Reducer。

代码4：CSVMapper.java

1. public class CSVMapper
2. extends Mapper<LongWritable, TextArrayWritable, Text, IntWritable> {
3. @Override
4. protected void map(LongWritable key, TextArrayWritable value, Context context)
5. throws IOException, InterruptedException {
6. String[] values = value.toStrings();
7. int sum = 0;
8. Text resultKey = new Text(values[0]);
9. for (int i = 1; i < values.length; i++) {
10. sum = sum + Integer.valueOf(values[i].trim());
11. }
12. IntWritable resultValue = new IntWritable(sum);
13. context.write(resultKey, resultValue);
14. }
15. }

在作业的提交部分，由于没有Reducer，所以将ReduceTask设置为了0

代码5：JustRun.java

1. public class JustRun extends Configured implements Tool{
2. @Override
3. public int run(String[] args) throws Exception {
4. Configuration conf = new Configuration();
6. Job job = new Job(conf);
7. job.setJobName("CSVTest");
8. job.setJarByClass(JustRun.class);
10. job.setMapperClass(CSVMapper.class);
12. job.setOutputKeyClass(Text.class);
13. job.setOutputValueClass(IntWritable.class);
14. job.setInputFormatClass(CSVInputFormat.class);
16. job.setNumReduceTasks(0);
18. FileInputFormat.setInputPaths(job, new Path(args[0]));
19. FileOutputFormat.setOutputPath(job, new Path(args[1]));
20. return job.waitForCompletion(true) 0 : 1;
21. }
22. public static void main(String[] args) throws Exception {
23. int ret = ToolRunner.run(new JustRun(), args);
24. System.exit(ret);
25. }
26. }

执行完毕后，输出如下，跟预想是一致的。

好了，这就是利用InputFormat对CSV文件的处理过程。除了CSV文件，还可根据处理数据的类型，写出更多的InputFormat。同时，我们还可以利用OutputFormat输出需要的格式。

转自

http://bukp.me/hadoop/work-with-csv-input-file-in-hadoop.html