Hadoop的SequenceFile写实例

 1 package com.mengyao.hadoop.hdfs;
 2 
 3 import java.io.File;
 4 import java.io.IOException;
 5 
 6 import org.apache.commons.io.FileUtils;
 7 import org.apache.hadoop.conf.Configuration;
 8 import org.apache.hadoop.fs.FileSystem;
 9 import org.apache.hadoop.fs.Path;
10 import org.apache.hadoop.io.IOUtils;
11 import org.apache.hadoop.io.IntWritable;
12 import org.apache.hadoop.io.SequenceFile;
13 import org.apache.hadoop.io.SequenceFile.Metadata;
14 import org.apache.hadoop.io.SequenceFile.Writer;
15 import org.apache.hadoop.io.Text;
16 import org.apache.hadoop.io.compress.DefaultCodec;
17 
18 /**
19  * 将本地文件以SequenceFile格式上传到HDFS上。
20  *
21  *
22  * SequeceFile是 Hadoop API提供的一种二进制文件支持。这种二进制文件直接将<key, value>对序列化到文件中。一般对小文件可以使用这种文件合并，即将文件名作为key，文件内容作为value序列化到大文件中。这种文件格式 有以下好处：
23  *         1)支持压缩，且可定制为基于Record或Block压缩（Block级压缩性能较优）
24  *         2)本地化任务支持：因为文件可以被切分，因此MapReduce任务时数据的本地化情况应该是非常好的。
25  *         3)难度低：因为是Hadoop框架提供的API，业务逻辑侧的修改比较简单。
26  *         坏处是需要一个合并文件的过程，且合并后的文件将不方便查看。
27 
28  * SequenceFile 是一个由二进制序列化过的key/value的字节流组成的文本存储文件，它可以在map/reduce过程中的input/output 的format时被使用。在map/reduce过程中，map处理文件的临时输出就是使用SequenceFile处理过的。
29  * SequenceFile分别提供了读、写、排序的操作类。SequenceFile的操作中有三种处理方式：
30  *         1） 不压缩数据直接存储。 //enum.NONE
31  *         2） 压缩value值不压缩key值存储的存储方式。//enum.RECORD
32  *         3）key/value值都压缩的方式存储。//enum.BLOCK
33  *
34  * 注意：存储在SequenceFile中的key和value不一定Hadoop的Writable序列化类型，只要是能被Java的Serializable序列化和反序列化的任意类型都可以。
35  *
36  * @author mengyao
37  *
38  */
39 public class SequenceFileWriter {
40     
41     public static void main(String[] args) throws IOException {
42         args = new String[]{"/usr/local/mapreduces/bookOutline.txt", "/mapreduces/seqfile/book1.txt"};
43         Configuration conf = new Configuration();
44         FileSystem fs = FileSystem.get(conf);
45         File inFile = new File(args[0]);
46         Path outputPath = new Path(args[1]);
47         IntWritable key = new IntWritable();
48         Text value = new Text();
49         SequenceFile.Writer writer = null;
50         try {
51             writer = SequenceFile.createWriter(conf,
52                     Writer.file(outputPath),
53                     Writer.keyClass(key.getClass()),
54                     Writer.valueClass(value.getClass()),
55                     Writer.bufferSize(fs.getConf().getInt("io.file.buffer.size", 4096)),
56                     Writer.replication(fs.getDefaultReplication(null)),
57                     Writer.blockSize(134217728),
58                     Writer.compression(SequenceFile.CompressionType.BLOCK, new DefaultCodec()),
59                     Writer.progressable(null),
60                     Writer.metadata(new Metadata()));
61             int count = 0;
62             for (String line : FileUtils.readLines(inFile)) {
63                 key.set(count++);
64                 value.set(line);
65                 writer.append(key, value);
66             }
67         } finally {
68             IOUtils.closeStream(writer);
69         }
70     }
71 }