圖片文件一般在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 = 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 = new Text();
while (reader.next(key, value)) {
fileutil.FileWriter("D://image3//"+System.currentTimeMillis()+".jpg", value.getBytes());
}
reader.close();
}

基於hbase+hdfs的小文件(圖片)存儲