Hadoop基于文件的数据结构

 

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);
	}
}