图片文件一般在100k一下，质量好一些的在几百k，特殊的图像可能达到10m左右，如果直接存储在hdfs上会对namenode的内存造成很大的压力，因为namenode的内存中会存储每个文件或者目录的inode信息。但是如果存储在hbase中，hbase的单个cell超过100k就会造成后期压力。因此采用hdfs的sequenceFile存储文件，hbase存储文件在sequenceFile中的索引信息。

sequenceFile存储

　　自hadoop2.7.3以后，sequenceFile的write有了一个新的参数：AppendIfExistsOption，实现了可以向现有的sequenceFile中追加数据。每次写入数据之前和之后都记录当前的拍偏移 writer.getLength()，分别记为startPos和endPos。

hbase索引存储

　　将文件名称(文件唯一标识)作为rowkey，存储该文件所在的sequenceFile的名称，以及存储时记录的startPos和endPos。

文件读取

　　　给定文件名，从hbase中检索到文件存储在hdfs对应的sequenceFile的name(包含路径)，以及在文件中的起止position。根据文件名和起止位置去hdfs读取文件内容。

代码如下：

    /**
     * 创建表
     */
    public static void createTable() throws IOException{
        HBaseAdmin admin = HbaseInit.getHBaseAdmin();       
        HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName));
        HColumnDescriptor fimalyDesc = new HColumnDescriptor("v");
        tableDesc.addFamily(fimalyDesc);       
        admin.createTable(tableDesc);
    }

　　/**
     * 将指定的文件写入文件系统，并将索引写入hbase
     */
    public static void FileWriter(byte[] image, int count) throws IOException {

        IntWritable key = new IntWritable();
        Text value = http://www.mamicode.com/new Text((byte[]) image);

        SequenceFile.Writer writer = null;
        Option optPath = SequenceFile.Writer.file(p);
        Option optKey = SequenceFile.Writer.keyClass(key.getClass());
        Option optVal = SequenceFile.Writer.valueClass(value.getClass());
        Option optExist = SequenceFile.Writer.appendIfExists(true);
        Option optCompress = SequenceFile.Writer.compression(CompressionType.RECORD);

        try {
            writer = SequenceFile.createWriter(fs.getConf(), optPath, optKey, optVal, optExist, optCompress);

            long time = System.currentTimeMillis();
            HTable table = HbaseInit.getTable(TableName.valueOf("imageIdx"));
            long startPos = 0;
            for(int i = 0; i<count; i++){                
                startPos = writer.getLength();
                writer.append(key, value);
                Put put = new Put(Bytes.toBytes(System.currentTimeMillis()+"/ddd/ddd.txt"));
                put.addColumn(Bytes.toBytes("v"), Bytes.toBytes("name"), Bytes.toBytes("/ddd/ddd.txt"));
                put.addColumn(Bytes.toBytes("v"), Bytes.toBytes("start"), Bytes.toBytes(startPos));
                put.addColumn(Bytes.toBytes("v"), Bytes.toBytes("end"), Bytes.toBytes(writer.getLength()));                   
                table.put(put);      
            }
            System.out.println("写入hdfs&hbase耗时： "+ (System.currentTimeMillis()-time));
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e);
        } finally {
            IOUtils.closeStream(writer);
        }
    }

/**
     * 读取指定rowkey对应的索引数据
     */
    public static HbaseStoryEntity GetIdx(String rowKey) throws IOException{        
        HTable table = HbaseInit.getTable(TableName.valueOf(tableName));
                
        Get get = new Get(Bytes.toBytes(rowKey));
        
        Result r = table.get(get);
        HbaseStoryEntity imageFile = new HbaseStoryEntity();
        for(Cell cell: r.listCells()){
            if(Bytes.toString(CellUtil.cloneQualifier(cell)).equals("name")){
                imageFile.setFileName(Bytes.toString(CellUtil.cloneValue(cell)));
            }else if(Bytes.toString(CellUtil.cloneQualifier(cell)).equals("start")){
                imageFile.setStartPos(Bytes.toLong(CellUtil.cloneValue(cell)));
            }else if(Bytes.toString(CellUtil.cloneQualifier(cell)).equals("end")){
                imageFile.setEndPos(Bytes.toLong(CellUtil.cloneValue(cell)));
            }
        }       
        return imageFile;
    }

/**
     * 按照索引数据从sequenceFileduq读取数据
     */
    public static void FileReader(HbaseStoryEntity imageIdx) throws IOException {

        SequenceFile.Reader.Option optlen = SequenceFile.Reader.length(imageIdx.getEndPos());
        SequenceFile.Reader reader = new SequenceFile.Reader(fs.getConf(),
                SequenceFile.Reader.file(new Path(imageIdx.getFileName())), optlen);
        reader.seek(imageIdx.getStartPos());
        IntWritable key = new IntWritable();
        Text value = http://www.mamicode.com/new Text();
        while (reader.next(key, value)) {
        fileutil.FileWriter("D://image3//"+System.currentTimeMillis()+".jpg", value.getBytes());
        }      
        reader.close();
    }

基于hbase+hdfs的小文件(图片)存储