spring hbase hbaseTemplate

spring hbase template延续spring 模板模式的一贯风格，只需要配置连接池属性和注册template，重写template的方法即可使用。

吐槽：个人当时网上搜索时，发现很多都是copy代码，用不了。无语。

开始之前：有一个可用的maven web工程，下载hbase client 和spring hadoop,spring hbase相关jar.

<!-- hbase begin -->
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>1.2.1</version>
<exclusions>
<exclusion>
<groupId>jdk.tools</groupId>
<artifactId>jdk.tools</artifactId>
</exclusion>
<exclusion>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>jdk.tools</groupId>
<artifactId>jdk.tools</artifactId>
<version>1.8</version>
<scope>system</scope>
<systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
</dependency>
<!-- hbase end -->

<!-- hadoop begin -->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>2.5.1</version>
</dependency>
<!-- hadoop end -->

<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-hadoop</artifactId>
<version>2.4.0.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-hadoop-core</artifactId>
<version>2.4.0.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase</artifactId>
<version>1.2.1</version>
<type>pom</type>
</dependency>

下面一一贴代码并解释：

一、配置template bean对象，并且设置hbase连接属性。

具体方法：在配置spring 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">
<!-- 配置zookeeper的信息，远程连接hbase时使用 -->
<hdp:configuration resources="classpath:/hbase-site.xml" />
<hdp:hbase-configuration configuration-ref="hadoopConfiguration" />
<!-- 配置HbaseTemplate -->
<bean id="hbaseTemplate" class="org.springframework.data.hadoop.hbase.HbaseTemplate">
<property name="configuration" ref="hbaseConfiguration">
</property>
<property name="encoding" value="UTF-8"></property>
</bean>
</beans>

说明：
1、hdp:configuration指定hadoop(hdfs)使用那个配置（可以直接指定，也可是是单独的文件）作为连接信息。

2、<hdp:hbase-configuration configuration-ref="hadoopConfiguration" /> 这句表明hbase的配置直接使用hadoop的配置。

3、<bean id="hbaseTemplate">注册hbaseTemplate bean。配置引用为2里的配置。

然后在相同文件夹下新建spring-hbase.xml(haddop config)所需的连接信文件hbase-site.xml.(这个文件便是hbase安装目录里的同名文件，根据自己的需求修改即可)

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- /** * * Licensed to the Apache Software Foundation (ASF) under one *
or more contributor license agreements. See the NOTICE file * distributed
with this work for additional information * regarding copyright ownership.
The ASF licenses this file * to you under the Apache License, Version 2.0
(the * "License"); you may not use this file except in compliance * with
the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software *
distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT
WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the
License for the specific language governing permissions and * limitations
under the License. */ -->
<configuration>
<property>
<name>hbase.rootdir</name>
<value>hdfs://192.168.1.2:9000/hbase</value>
</property>
<property>
<name>hbase.cluster.distributed</name>
<value>true</value>
</property>
<property>
<name>hbase.master.port</name>
<value>16000</value>
</property>
<property>
<name>hbase.master.info.port</name>
<value>60010</value>
</property>
<property>
<name>hbase.zookeeper.quorum</name>
<value>192.168.1.2,192.168.1.3</value>
</property>
<property>
<name>hbase.zookeeper.property.dataDir</name>
<value>/usr/zookeeper/data</value>
</property>
<property>
<name>hbase.zookeeper.property.clientPort</name>
<value>2181</value>
</property>
<property>
<name>fs.hdfs.impl</name>
<value>org.apache.hadoop.hdfs.DistributedFileSystem</value>
</property>
</configuration>

上面的文件定义了连接的hdfs入口（hbase配置hdfs所使用的路径，需要手动建立hdfs路径。端口为hdfs访问端口，不修改为默认9000），分布式类型（单点false,伪分布式和分布式均为true），主节点端口(默认16000)，页面访问端口（默认60010），zookeper集群、端口、数据文件存放位置。
配置中机器使用ip还是机器名需要根据hbase的安装配置。如果配置机器名，就使用机器名；否则，使用IP即可。不过，使用机器名时，需要在本地添加host认证。（具体问百度）

tips: 如果是伪分布式，去掉zookeper data文件配置，将主机设为单点地址（hbase安装机器ip或机器名），端口默认（或按hbase安装配置）。

二、配置完文件后（记得spring 扫描spring-hbase.xml）,编写template调用工具类。

为了增加通用性，我自己增加两个实体类作为辅助对象。

1、HQuery hbase查询时所使用对象，主要申明查询的startrow,endrow,scan,filter等。以下为片段

@Component
public class HQuery {

private String table;
private String family;
private String qualifier;
private String qualifierValue;
private String row;
private String startRow;
private String stopRow;
private Filter filter;
private PageFilter pageFilter;
private Scan scan;

private List<HBaseColumn> columns = Lists.newArrayList();

2、HBaseColumn 数据多字段写入时的辅助实体。family，字段名和值
public class HBaseColumn {

private String family;
private String qualifier;
private String value;

3、HBaseTemplate 工具类。
@Repository
public class HBaseTemplate {

private final Logger logger = Logger.getL
4000
ogger(this.getClass());

ApplicationContext context = new ClassPathXmlApplicationContext(new String[] { "spring-hbase.xml" });

// BeanFactory factory = (BeanFactory) context;
HbaseTemplate htemplate = (HbaseTemplate) context.getBean("hbaseTemplate");

/**
* 写数据
*
* @param tableName
* @param action
* @return
*/
public Object execute(HQuery query) {
if(StringUtils.isBlank(query.getRow()) || query.getColumns().isEmpty()){
return null;
}
return htemplate.execute(query.getTable(), new TableCallback<Object>() {
@SuppressWarnings("deprecation")
@Override
public Object doInTable(HTableInterface table) throws Throwable {
try {
byte[] rowkey = query.getRow().getBytes();
Put put = new Put(rowkey);
for(HBaseColumn col:query.getColumns()){
put.addColumn(Bytes.toBytes(col.getFamily()), Bytes.toBytes(col.getQualifier()),
Bytes.toBytes(col.getValue()));
}
table.put(put);
} catch (Exception e) {
logger.warn("==> hbase get object fail> "+query.getRow());
}
return null;
}

});
}

/**
* 通过表名和key获取一行数据
*
* @param tableName
* @param rowName
* @return
*/
public <T> T get(HQuery query, Class<T> c) {

if(StringUtils.isBlank(query.getTable()) || StringUtils.isBlank(query.getRow())){
return null;
}

return htemplate.get(query.getTable(), query.getRow(), new RowMapper<T>() {
public T mapRow(Result result, int rowNum) throws Exception {
List<Cell> ceList = result.listCells();
T item=c.newInstance();
JSONObject obj = new JSONObject();
if (ceList != null && ceList.size() > 0) {
for (Cell cell : ceList) {
obj.put(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(),
cell.getQualifierLength()),
Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
}
}else{
return null;
}
item=JSON.parseObject(obj.toJSONString(), c);
return item;
}
});
}

/**
* 通过表名 key 和 列族 和列 获取一个数据
*
* @param tableName
* @param rowName
* @param familyName
* @param qualifier
* @return
*/
public String getColumn(HQuery query) {

if(StringUtils.isBlank(query.getTable()) || StringUtils.isBlank(query.getRow())
|| StringUtils.isBlank(query.getFamily()) || StringUtils.isBlank(query.getQualifier())){
return null;
}

return htemplate.get(query.getTable(), query.getRow(), query.getFamily(), query.getQualifier(), 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 HQuery
* @return
*/
public <T> List<T> find(HQuery query,Class<T> c) {
//如果未设置scan,设置scan
if (query.getScan() == null) {

//起止搜索
if(StringUtils.isNotBlank(query.getStartRow()) && StringUtils.isNotBlank(query.getStopRow())){
query.setSearchLimit(query.getStartRow(), query.getStopRow());
}

//主要配合pageFilter，指定起始点
if(StringUtils.isNotBlank(query.getStartRow())){
query.setScanStartRow(query.getStartRow());
}

//列匹配搜索
if(StringUtils.isNotBlank(query.getFamily()) &&StringUtils.isNotBlank(query.getQualifier())
&&StringUtils.isNotBlank(query.getQualifierValue())){
query.setSearchEqualFilter(query.getFamily(),query.getQualifier(),query.getQualifierValue());
}

//分页搜索
if(query.getPageFilter()!=null){
query.setFilters(query.getPageFilter());
}

if(query.getScan()==null){
query.setScan(new Scan());
}
}

//设置缓存
query.getScan().setCacheBlocks(false);
query.getScan().setCaching(2000);

return htemplate.find(query.getTable(), query.getScan(), new RowMapper<T>() {
@Override
public T mapRow(Result result, int rowNum) throws Exception {

List<Cell> ceList = result.listCells();
JSONObject obj = new JSONObject();
T item =c.newInstance();
if (ceList != null && ceList.size() > 0) {
for (Cell cell : ceList) {
// String row = Bytes.toString(cell.getRowArray(),
// cell.getRowOffset(), cell.getRowLength());
// String family = Bytes.toString(cell.getFamilyArray(),
// cell.getFamilyOffset(),
// cell.getFamilyLength());

String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(),
cell.getValueLength());

String quali = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(),
cell.getQualifierLength());
if(value.startsWith("[")){
obj.put(quali, JSONArray.parseArray(value));
}else{
obj.put(quali, value);
}
}
}
item =JSON.parseObject(obj.toJSONString(), c);
return item;
}

});
}

public void delete(HQuery query){
htemplate.delete(query.getTable(), query.getRow(), query.getFamily());
}
}

以上几个方法均是hbaseTemplate的方法扩展，个人只需要实现回调方法逻辑处理内容（callback或rowmapper）即可。
为了增加查询效率。在scan中设置了缓存。

query.getScan().setCacheBlocks(false);

query.getScan().setCaching(2000);

以上，均为个人实际代码。转载请注明出处。----------------------------------