Hadoop基于文件的数据结构

zeromax

　　1 SequenceFile的理解 

  （1）SequenceFile是Hadoop用来存储二进制形式的对而设计的一种平面文件（Flat File）；
（2）可以把SequenceFile当做一个容器，把所有文件打包到SequenceFile类中可以高效的对小文件进行存储和处理；
（3）SequenceFile文件并不按照其存储的key进行排序存储，SequenceFile的内部类Writer提供了append功能；
（4）SequenceFile中的key和value可以是任意类型Writable或者是自定义的Writable类型；
　　2 SequenceFile:支持数据压缩（记录压缩，块压缩）

    a.无压缩类型：如果没有启用压缩（默认设置），那么每个记录就由他的记录长度（字节数）、键的长度、
键和值组成。长度字段为四字节。
b.记录压缩类型：记录压缩格式与无压缩格式基本相同，不同的是值字节是用定义在头部的编码器来压缩的。
注意，键是不压缩的。
c.块压缩类型：块压缩一次压缩多个记录，因此它比记录压缩更紧凑，而且一般优先选择。
当记录的字节数达到最小大小，才会添加到块。改最小值由io.seqfile.compress.blocksize中的属性定义。默认值是1000000字节。格式为记录数、键长度、键、值长度、值。
　　3 优缺点
　　缺点：需要一个合并文件过程，并且合并后的文件不方便查看 
　　SequenceFile写过程：

（1）创建Configuration；（2）获取FileSystem；（3）创建文件输出路径Path；
（4）调用SequenceFile.createWriter得到SequenceFile.Writer对象；
（5）调用SequenceFile.Writer.append追加写入文件；（6）关闭流；
　　SequenceFile读过程：

（1）创建Configuration；（2）获取FileSystem；（3）创建文件输出路径Path；
（4）new一个SequenceFile Reader进行读取；（5）得到keyClass和valueClass；（6）关闭流；
　　二、MapFile
　　（1）MapFile是经过排序的索引的SequenceFile，可以根据key进行查找；
　　（2）MapFile的key 是WritableComparable类型的，而value是Writable类型的；
　　（3）可以使用MapFile.fix()方法来重建索引，把SequenceFile转换成MapFile
　　（4）它有两个静态成员变量：
　　static String DATA_FILE_NAME //数据文件名
　　static String INDEX_FILE_NAME //索引文件名
　　三、实例代码
　　1 SequentceFile 的读写 

package org.tony.file;
public class SequenceFileWriter {
public static String uri = "hdfs://192.168.102.136:9000";
public static String[] data = { "one,two", "three,four", "five,six",
"seven,eight", "night,ten" };
public static void main(String[] args) throws IOException {
write();
read();
}
/**
* @Title: write
* @Description: （1）创建Configuration；（2）获取FileSystem；（3）创建文件输出路径Path；
（4）调用SequenceFile.createWriter得到SequenceFile.Writer对象；
（5）调用SequenceFile.Writer.append追加写入文件；（6）关闭流；
* @return void    返回类型
* @throws
*/
public static void write() throws IOException {
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
Path path = new Path("/tmp.seq");
// Path path = new Path("/tmp1.seq"); 采用压缩
IntWritable key = new IntWritable();
Text value = new Text();
SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, path,
key.getClass(), value.getClass());
// SequenceFile.Writer writer =
// SequenceFile.createWriter(fs,conf,path,key.getClass(),value.getClass(),CompressionType.RECORD,new
// BZip2Codec()); 采用压缩，用BZip2压缩算法
for (int i = 0; i < 100; i++) { // 写入100次
key.set(100 - i);
value.set(data[i % data.length]);
writer.append(key, value);
}
IOUtils.closeStream(writer);
}
/**
* @Title: read
* @Description: （1）创建Configuration；（2）获取FileSystem；（3）创建文件输出路径Path；
（4）new一个SequenceFile Reader进行读取；（5）得到keyClass和valueClass；（6）关闭流；
* @return void    返回类型
* @throws
*/
public static void read() throws IOException {
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
Path path = new Path("/tmp.seq");
// Path path = new Path("/tmp1.seq"); 读取压缩文件
SequenceFile.Reader reader = new SequenceFile.Reader(fs, path, conf);
Writable key = (Writable) ReflectionUtils.newInstance(
reader.getKeyClass(), conf);
Writable value = (Writable) ReflectionUtils.newInstance(
reader.getValueClass(), conf);
while (reader.next(key, value)) {
System.out.println("key = " + key);
System.out.println("value = " + value);
System.out.println("position = " + reader.getPosition());
}
IOUtils.closeStream(reader);
}
}

　　2 MapFile使用

package org.tony.file;
public class MapFileTest {
public static String uri = "hdfs://192.168.102.136:9000"; //访问hdfs 的uri
public static String[] data = { "one,two", "three,four", "five,six",
"seven,eight", "night,ten" };
public static void main(String[] args) throws Exception {
//write();
//read();
seqToMapFile();
}
/**用Mapfile写文件
* @Title: write
* @Description: （1）创建Configuration；（2）获取FileSystem；（3）创建文件输出路径Path；
（4）new一个MapFile.Writer对象；（5）调用MapFile.Writer.append追加写入文件；（6）关闭流；
* @return void    返回类型
* @throws
*/
public static void write() throws IOException {
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
Path path = new Path("/tmpdata.map");
IntWritable key = new IntWritable();
Text value = new Text();
MapFile.Writer writer = new MapFile.Writer(conf, fs, path.toString(),
key.getClass(), value.getClass());
for (int i = 0; i < 100; i++) {
key.set(i + 1);
value.set(data[i % data.length]);
writer.append(key, value);
}
IOUtils.closeStream(writer);
}
/**用Mapfile读文件
* @Title: read
* @Description: （1）创建Configuration；（2）获取FileSystem；（3）创建文件输出路径Path；
（4）new一个MapFile.Reader对象；（5）得到keyClass和valueClass；（6）关闭流；
* @return void    返回类型
* @throws
*/
public static void read() throws IOException {
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
Path path = new Path("/tmpdata.map");
MapFile.Reader reader = new MapFile.Reader(fs, path.toString(), conf);
WritableComparable key = (WritableComparable) ReflectionUtils
.newInstance(reader.getKeyClass(), conf);
Writable value = (Writable) ReflectionUtils.newInstance(
reader.getValueClass(), conf);
while (reader.next(key, value)) {
System.out.println("key = " + key);
System.out.println("value = " + value);
}
IOUtils.closeStream(reader);
}
/**将sequence文件转换为MapFile文件
* @Title: seqToMapFile
* @Description: TODO
* @return void    返回类型
* @throws
* 1 创建tmp1.map文件夹
* 2 复制tmp1.seq  SequenceFile文件到tmp1.map文件夹下面，并重命名data $./hadoop fs -mv /tmp1.seq /tmp1.map/data
* 3 运行程序
*/
public static void seqToMapFile() throws Exception{
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
Path map = new Path("/tmp1.map"); //文件夹位置
//MapFile.DATA_FILE_NAME 为seq文件移动到tmp1.map文件夹下面的文件名称
Path mapData = new Path(map,MapFile.DATA_FILE_NAME);
SequenceFile.Reader reader = new SequenceFile.Reader(fs,mapData,conf);
Class key = reader.getKeyClass();
Class value = reader.getValueClass();
reader.close();
long entries = MapFile.fix(fs, map, key, value, false, conf);
System.out.printf("Created MapFile %s with %d entries\n",map,entries);
}
}