java 获取 hbase数据 springdatahadoop -- hbasetemplate

本是古典何须时尚 2022-08-06 10:28 137阅读 0赞

**java 利用 ** **sping-data-hadoop       HbaseTemplate 操作hbase**

--------------------

转载请注明出处： [http://blog.csdn.net/linlinv3/article/details/42737113][http_blog.csdn.net_linlinv3_article_details_42737113]

官网api地址： [http://docs.spring.io/autorepo/docs/spring-data-hadoop/2.1.0.M1-hadoop24/api/org/springframework/data/hadoop/hbase/HbaseTemplate.html][http_docs.spring.io_autorepo_docs_spring-data-hadoop_2.1.0.M1-hadoop24_api_org_springframework_data_hadoop_hbase_HbaseTemplate.html]

--------------------

为了让自己的项目能够融合hbase ，这里讲解了 如何 利用hbasetemplate 对hbase进行操作

具体的hbase的特性和原理就不多说了

直接贴程序

**配置文件**

1、建立 spring-hbase.xml 获取连接池

第一种：直接在配置文件注明 hdfs的端口和zk的地址以及端口进行连接

<?xml version="1.0" encoding="UTF-8"?>
    	<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:hdp="http://www.springframework.org/schema/hadoop"
       xmlns:beans="http://www.springframework.org/schema/beans"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="
        http://www.springframework.org/schema/beans 
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/hadoop 
        http://www.springframework.org/schema/hadoop/spring-hadoop.xsd
        http://www.springframework.org/schema/context 
        http://www.springframework.org/schema/context/spring-context-3.1.xsd">
       
       <context:property-placeholder location="hbase.properties" />
       
          
    	<bean id="htemplate" class="org.springframework.data.hadoop.hbase.HbaseTemplate">  
    		  <property name="configuration" ref="hbaseConfiguration">
    		  </property>
    	 </bean>  
         
    	 <hdp:configuration>
    	   	  fs.default.name="hdfs://192.168.0.173:8082"
    	 </hdp:configuration> 
    	 
    	 <hdp:hbase-configuration zk-quorum="192.168.0.173" zk-port="2181" /> 
     </beans>

第二种：只需要配置zk的地址和端口，但是需要在classpath另外新增 hbase-site.xml的配置

spring-hbase.xml 配置：

<?xml version="1.0" encoding="UTF-8"?>
    	<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:hdp="http://www.springframework.org/schema/hadoop"
       xmlns:beans="http://www.springframework.org/schema/beans"
       xsi:schemaLocation="
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/hadoop http://www.springframework.org/schema/hadoop/spring-hadoop.xsd">
          
    	<hdp:configuration resources="classpath:/hbase-site.xml" />
    	<hdp:hbase-configuration configuration-ref="hadoopConfiguration" />
    	
    	<bean id="htemplate" class="org.springframework.data.hadoop.hbase.HbaseTemplate">
    		<property name="configuration" ref="hbaseConfiguration">
    		</property>
    	<property name="encoding" value="UTF-8"></property>
    	</bean>
     </beans>

hbase-site.xml 配置  放在classpath下（直接扔到src下即可）

<?xml version="1.0"?>
    <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
    <configuration>
    	<property>
    		<name>hbase.zookeeper.quorum</name>
    		<value>192.168.0.173</value>
    	</property>
    
    	<property>
    		<name>hbase.zookeeper.property.clientPort</name>
    		<value>2181</value>
    	</property>
    </configuration>

配置完成

--------------------

**程序代码**

官网的api地址已经在上面发过了，这里只做了一个简单的demo，列举了简单的 get  find execue 方法

package cn.fulong.hbase;
    
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    import org.apache.hadoop.hbase.Cell;
    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.client.Scan;
    import org.apache.hadoop.hbase.util.Bytes;
    import org.springframework.beans.factory.BeanFactory;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    import org.springframework.data.hadoop.hbase.HbaseTemplate;
    import org.springframework.data.hadoop.hbase.RowMapper;
    import org.springframework.data.hadoop.hbase.TableCallback;
    
    import cn.fulong.view.HbaseModel;
    public class HbaseTest {
     	//ApplicationContext context = new ClassPathXmlApplicationContext(new String[] { "application_hbase.xml" });  
     	ApplicationContext context = new ClassPathXmlApplicationContext(new String[] { "spring_hbase.xml" });  
    
    	BeanFactory factory = (BeanFactory) context; 
    	HbaseTemplate htemplate = (HbaseTemplate) factory.getBean("htemplate");
    	Map hMap = new HashMap<String, List<HbaseModel>>();
    	public String key;
    	public String familyName ;
    	public String qualifier;
    	public String value;
    		
    	public String getValue() {
    		return value;
    	}
    	public void setValue(String value) {
    		this.value = value;
    	}
    	public String getFamilyName() {
    		return familyName;
    	}
    	public void setFamilyName(String familyName) {
    		this.familyName = familyName;
    	}
    	public String getQualifier() {
    		return qualifier;
    	}
    	public void setQualifier(String qualifier) {
    		this.qualifier = qualifier;
    	}
    	public String getKey() {
    		return key;
    	}
    	public void setKey(String key) {
    		this.key = key;
    	}
    	
    		 
    		
    	public static void main(String[] args) {
    		//PrefixFilter
    		HbaseTest h = new HbaseTest();
     
    		for(int i=0;i<=10000;i++){
    			h.setKey("linlin"+i);
    			h.setFamilyName("info"); 
    			h.setQualifier("service");
    			h.setValue(i+"技术创新和质量服务");
    			h.execute("linlintest", null);
    		}
    	
    	 List<Map<String,Object>>  mapList1 = h.find("linlintest",null,null);
    	 	System.out.println("2");
    	}
    	/**
    	 * 写数据
    	 * @param tableName
    	 * @param action
    	 * @return
    	 */
    	public Boolean execute(String tableName, TableCallback<Boolean> action) {  
    		return htemplate.execute(tableName, new TableCallback<Boolean>() {
                public Boolean doInTable(HTableInterface table) throws Throwable {
                    boolean flag = false;
                    try{
                    	byte[] rowkey = key.getBytes();
                    	Put put = new Put(rowkey);
                    	put.add(Bytes.toBytes(familyName),Bytes.toBytes(qualifier), Bytes.toBytes(value));
                    	table.put(put);
                     flag = true;
                    }catch(Exception e){
                        e.printStackTrace();
                    }
                    return flag;
                }
            });
        }  
    	 /**
    	  * 通过表名和key获取一行数据
    	  * @param tableName
    	  * @param rowName
    	  * @return
    	  */
    	public Map<String, Object> get(String tableName, String rowName) {
    		 return htemplate.get(tableName, rowName,new RowMapper<Map<String,Object>>(){
    	           public Map<String,Object> mapRow(Result result, int rowNum) throws Exception {	   
    	        	   List<Cell> ceList =   result.listCells();
    	        	   Map<String,Object> map = new HashMap<String, Object>();
    	        	   		if(ceList!=null&&ceList.size()>0){
    	        	   			for(Cell cell:ceList){
    	        	   				map.put(Bytes.toString(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength())+
    	        					   "_"+Bytes.toString( cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()), 
    	        					   Bytes.toString( cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
    	        		   }
    	        	   }
    					return  map;
    	           }
    		     });
    	}
    	
    	/**
    	 * 通过表名  key 和 列族 和列 获取一个数据
    	 * @param tableName
    	 * @param rowName
    	 * @param familyName
    	 * @param qualifier
    	 * @return
    	 */
    	public String get(String tableName ,String rowName, String familyName, String qualifier) {
    		  return htemplate.get(tableName, rowName,familyName,qualifier ,new RowMapper<String>(){
    		         public String mapRow(Result result, int rowNum) throws Exception {	  
    		        	 List<Cell> ceList =   result.listCells();
    		        	 String res = "";
    		        	 if(ceList!=null&&ceList.size()>0){
    		        		 for(Cell cell:ceList){
    		        			 res = Bytes.toString( cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
    		        		 }
    		        	 }
    		      	   return res;
    		         }
    		  });
    	}
    	
    	
    	 /**
    	  * 通过表名，开始行键和结束行键获取数据
    	  * @param tableName
    	  * @param startRow
    	  * @param stopRow
    	  * @return
    	  */
    	public List<Map<String,Object>> find(String tableName , String startRow,String stopRow) {
    		 Scan scan = new Scan();
    		 if(startRow==null){
    			 startRow="";
    		 }
    		 if(stopRow==null){
    			 stopRow="";	
    		 }
    		 scan.setStartRow(Bytes.toBytes(startRow));
    		 scan.setStopRow(Bytes.toBytes(stopRow));
    		/* PageFilter filter = new PageFilter(5);
    		 scan.setFilter(filter);*/
    		 return 	htemplate.find(tableName, scan,new RowMapper<Map<String,Object>>(){
    	           public Map<String,Object> mapRow(Result result, int rowNum) throws Exception {	
    	        	  
    		        	 List<Cell> ceList =   result.listCells();
    		        	 Map<String,Object> map = new HashMap<String,Object>();
    		        	 Map<String,Map<String,Object>> returnMap = new HashMap<String,Map<String,Object>>();
    		        	 String  row = "";
    		        	 if(ceList!=null&&ceList.size()>0){
    		        		   for(Cell cell:ceList){
    							row =Bytes.toString( cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
    							String value =Bytes.toString( cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
    							String family =  Bytes.toString(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength());
    							String quali = Bytes.toString( cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength());
    							map.put(family+"_"+quali, value);
    		        		   }
    		        		   map.put("row",row );
     		        	   }
    		        	   return  map;
    		           }
    			     });
    	}
    
    	
    	/* public  void scanValueByFilter(String tableName,String row) throws IOException{
             HTable table = new HTable(conf, tableName);  
             Scan scan = new Scan();
             scan.setFilter(new PrefixFilter(row.getBytes()));
             ResultScanner resultScanner = table.getScanner(scan);
             for(Result rs:resultScanner){
        
                 
             }
             
         }*/
    }

--------------------

**下载**

具体可运行代码下载地址    [http://download.csdn.net/detail/linlinv3/8580185][http_download.csdn.net_detail_linlinv3_8580185]

项目目录结构和所需jar包如下图所示：

![Center][]

[http_blog.csdn.net_linlinv3_article_details_42737113]: http://write.blog.csdn.net/postedit/42737113
[http_docs.spring.io_autorepo_docs_spring-data-hadoop_2.1.0.M1-hadoop24_api_org_springframework_data_hadoop_hbase_HbaseTemplate.html]: http://docs.spring.io/autorepo/docs/spring-data-hadoop/2.1.0.M1-hadoop24/api/org/springframework/data/hadoop/hbase/HbaseTemplate.html
[http_download.csdn.net_detail_linlinv3_8580185]: http://download.csdn.net/detail/linlinv3/8580185
[Center]: /images/20220806/c22ef242614e40179ee65df5d248b8a6.png