Hadoop的SequenceFile写实例

package com.mengyao.hadoop.hdfs;

import java.io.File;
import java.io.IOException;

import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.Metadata;
import org.apache.hadoop.io.SequenceFile.Writer;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.DefaultCodec;

/**
* 将本地文件以SequenceFile格式上传到HDFS上。
*
*
* SequeceFile是 Hadoop API提供的一种二进制文件支持。这种二进制文件直接将<key, value>对序列化到文件中。一般对小文件可以使用这种文件合并，即将文件名作为key，文件内容作为value序列化到大文件中。这种文件格式 有以下好处：
*         1)支持压缩，且可定制为基于Record或Block压缩（Block级压缩性能较优）
*         2)本地化任务支持：因为文件可以被切分，因此MapReduce任务时数据的本地化情况应该是非常好的。
*         3)难度低：因为是Hadoop框架提供的API，业务逻辑侧的修改比较简单。
*         坏处是需要一个合并文件的过程，且合并后的文件将不方便查看。

* SequenceFile 是一个由二进制序列化过的key/value的字节流组成的文本存储文件，它可以在map/reduce过程中的input/output 的format时被使用。在map/reduce过程中，map处理文件的临时输出就是使用SequenceFile处理过的。
* SequenceFile分别提供了读、写、排序的操作类。SequenceFile的操作中有三种处理方式：
*         1） 不压缩数据直接存储。 //enum.NONE
*         2） 压缩value值不压缩key值存储的存储方式。//enum.RECORD
*         3）key/value值都压缩的方式存储。//enum.BLOCK
*
* 注意：存储在SequenceFile中的key和value不一定Hadoop的Writable序列化类型，只要是能被Java的Serializable序列化和反序列化的任意类型都可以。
*
* @author mengyao
*
*/
public class SequenceFileWriter {

public static void main(String[] args) throws IOException {
args = new String[]{"/usr/local/mapreduces/bookOutline.txt", "/mapreduces/seqfile/book1.txt"};
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);
File inFile = new File(args[0]);
Path outputPath = new Path(args[1]);
IntWritable key = new IntWritable();
Text value = new Text();
SequenceFile.Writer writer = null;
try {
writer = SequenceFile.createWriter(conf,
Writer.file(outputPath),
Writer.keyClass(key.getClass()),
Writer.valueClass(value.getClass()),
Writer.bufferSize(fs.getConf().getInt("io.file.buffer.size", 4096)),
Writer.replication(fs.getDefaultReplication(null)),
Writer.blockSize(134217728),
Writer.compression(SequenceFile.CompressionType.BLOCK, new DefaultCodec()),
Writer.progressable(null),
Writer.metadata(new Metadata()));
int count = 0;
for (String line : FileUtils.readLines(inFile)) {
key.set(count++);
value.set(line);
writer.append(key, value);
}
} finally {
IOUtils.closeStream(writer);
}
}
}