hbase 上传下载文件

短命女 2022-08-05 15:03 57阅读 0赞

hbase 是高读的，虽然写的速度不很乐观，但是有时候也需要往里面存一些文件。（一般建议存放在hdfs上），这里讲一下怎么把文件存储到hbase上

首先大家都知道的，hbase 只支持 byte 的存储，所以我们首先要做的是吧文件变换为byte

以下就用代码来描述这些：

这里用了spring-data 的hbasetemplate

具体配置可以看上篇文章[hbase 相关配置][hbase]

然后是java代码

首先，是吧文件转换为流 在转换为字节的代码

import java.io.ByteArrayInputStream;
    import java.io.ByteArrayOutputStream;
    import java.io.File;
    import java.io.FileInputStream;
    import java.io.IOException;
    import java.io.InputStream;
    import java.util.Map;
    
    import javax.annotation.Resource;
    
    import org.apache.hadoop.hbase.client.HTableInterface;
    import org.apache.hadoop.hbase.client.Put;
    import org.apache.hadoop.hbase.client.Result;
    import org.apache.hadoop.hbase.util.Bytes;
    import org.springframework.data.hadoop.hbase.HbaseTemplate;
    import org.springframework.data.hadoop.hbase.RowMapper;
    import org.springframework.data.hadoop.hbase.TableCallback;

这里是自己定义的变量：

private String fileFileName;
    	private File file;
    	private String downloadFileName;
    	@Resource(name = "htemplate")
    	private  HbaseTemplate htemplate ;//= (BaseHbase) Platform.getInstance().getBean("baseHbaseImpl");
    
    	public HbaseTemplate getHtemplate() {
    		return htemplate;
    	}
    	public void setHtemplate(HbaseTemplate htemplate) {
    		this.htemplate = htemplate;
    	}
    	public File getFile() {
    		return file;
    	}
    	public void setFile(File file) {
    		this.file = file;
    	}
    	public String tableName;
    	/**
    	 * 行健
    	 */
    	public String key;
    	/**
    	 * 列族
    	 */
    	public String familyName;
    	/**
    	 * 存放图片的列
    	 */
    	public String imageQualifier;
    	/**
    	 * 存放图片名称的列
    	 */
    	public String imageNameQualifier;
    	public byte[] value;

/**
    	 * 文件流转换为字节
    	 * @return
    	 * @throws IOException
    	 */
    	public byte[]  getSource() throws IOException{  
    		 int bufLength =  (int) file.length();  
    	     FileInputStream in = new FileInputStream(file);  
    	     byte[] buf = new byte[bufLength];
    	     int readLen;
    	     ByteArrayOutputStream byout = new ByteArrayOutputStream();
    		 while ((readLen = in.read(buf, 0, bufLength)) > 0) {
    			byout.write(buf, 0, readLen);
    		 }
    		in.close();
    		byte[]  byteContent = byout.toByteArray(); 
    		return byteContent;
    	}

============================================================================

然后调用 hbaseTemplate （spring-data）实现对habse 的操作

/**
    	 * 写数据
    	 * @param tableName
    	 * @param action
    	 * @return
    	 * @throws IOException 
    	 */
    	public String uploadFile() throws IOException { 
    		
    		 value = getSource();
    		 Boolean bo =  this.execute(tableName,Bytes.toBytes(key) , Bytes.toBytes(familyName), Bytes.toBytes(imageQualifier) , value, null);//存储图片
    		 Boolean b1 =  this.execute(tableName,Bytes.toBytes(key) , Bytes.toBytes(familyName), Bytes.toBytes(imageNameQualifier) , Bytes.toBytes(fileFileName) , null);//存储图片名称
    		 if(bo){
    			 return "success";
    		 }else{
    			 return "error";
    		 }
    	}

/**
    	 * 
    	 * @param tableName  表名
    	 * @param key  行健
    	 * @param familyName 列族
    	 * @param qualifier 列
    	 * @param value 值
    	 * @param action 
    	 * @return
    	 */
    	private Boolean execute(String tableName, final byte[] key ,final  byte[] familyName ,final byte[] qualifier,final byte[] value,  TableCallback<Boolean> action) {  
    		return htemplate.execute(tableName, new TableCallback<Boolean>() {
                public Boolean doInTable(HTableInterface table) throws Throwable {
                    boolean flag = false;
                    try{
                    	Put put = new Put(key);
                    	put.add(familyName,qualifier,value);
                    	table.put(put);
                        flag = true;
                    }catch(Exception e){
                        e.printStackTrace();
                    }
                    return flag;
                }
            });
        }

然后是下载，用的struts，直接调用getInputStream的方法

/**
    	 * 下载hdfs路径上文件
    	 * @return
    	 */
    	public String downloadFile(){
    		//fs.copyFromLocalFile(new Path(local), new Path(remote));
    		return "success";
    		
    	}
    	public InputStream getInputStream() throws IOException {
    		/*downloadFileName =  getPicName();
    		byte[] data = getBytes();*/
    	 
    		downloadFileName = Bytes.toString(this.getPicName(tableName, key));
    		byte[] data = this.getPic(tableName, key);
    		try {
    			InputStream is = new ByteArrayInputStream(data); 
     			 return is;
    		} catch (Exception e) {
    			e.printStackTrace();
    			return null;
    		} 
    	}
    	
    	/**
    	 * 从hbase获取图片 byte
    	 * @param tableName
    	 * @param rowName
    	 * @return
    	 */
    	public byte[]  getPic(String tableName, String rowName) {
    			//byte[]  by = "";
    		    return 	htemplate.get(tableName, rowName, new RowMapper<byte[]>(){
    				@Override
    				public byte[] mapRow(Result result, int rowNum) throws Exception {
    					byte[] by = result.getValue(Bytes.toBytes(familyName), Bytes.toBytes(imageQualifier));
    					//byte[] by = Bytes.toString();
    		 			return by;
    				}
    		 	});
    		  
    	}
    	/**
    	 * 获取图片name
    	 * @param tableName
    	 * @param rowName
    	 * @return
    	 */
    	public byte[]  getPicName(String tableName, String rowName) {
    			//byte[]  by = "";
    		    return 	htemplate.get(tableName, rowName, new RowMapper<byte[]>(){
    				@Override
    				public byte[] mapRow(Result result, int rowNum) throws Exception {
    					byte[] by = result.getValue(Bytes.toBytes(familyName), Bytes.toBytes(imageNameQualifier));
    					//byte[] by = Bytes.toString();
    		 			return by;
    				}
    		 	});
    		  
    	}

需要的jar： 这里是spring-data的所有依赖包。struts的和spring 的其他的需要自己去整合了。

![Center][]

[hbase]: http://blog.csdn.net/linlinv3/article/details/42737113
[Center]: /images/20220805/1e9fbdb28b804094b4a07f237df20ea5.png