spring data hadoop 访问hbase

本文介绍spring boot 集成 spring data hadoop 以访问hbase 。

版本：

spring boot ： 2.0.4.RELEASE
spring data hadoop ： 2.5.0.RELEASE
hbase client ： 1.2.6

spring data hadoop 介绍

Spring for Apache Hadoop提供与Spring Framework的集成，以创建和运行Hadoop MapReduce，Hive和Pig作业，以及使用HDFS和HBase。

注意： Spring for Apache Hadoop项目将于2019年4月5日终止。我们将根据需要发布2.5.x维护版本，直到那时为止，然后将项目移至attic。 当前的Apache Hadoop 2.5.0版本是使用Apache Hadoop 2.7.3版本构建的，应该与最流行的Hadoop发行版的最新版本兼容。

功能：

• 支持创建使用依赖注入配置的Hadoop应用程序，并作为标准Java应用程序运行，而不是使用Hadoop命令行程序。

• 与Spring Boot集成，可以简单地创建连接到HDFS以读取和写入数据的Spring应用程序。

• 创建和配置使用Java MapReduce，Streaming，Hive，Pig或HBase的应用程序

• Spring Batch的扩展，支持为任何类型的Hadoop作业或HDFS操作创建基于Hadoop的工作流。

• 使用任何基于JVM的脚本语言编写HDFS操作脚本。

• 轻松创建基于Spring Boot的应用程序，可以部署在YARN上执行。

• HBase的DAO支持（模板和回调）。

• 支持Hadoop安全性。

Spring for Apache Hadoop支持许多Apache版本以及Pivotal，Hortonworks和Cloudera的商业发行版。

maven 配置

<dependencies>
        <!-- spring data hadoop  -->
        <dependency>
            <groupId>org.springframework.data</groupId>
            <artifactId>spring-data-hadoop</artifactId>
            <version>2.5.0.RELEASE</version>
        </dependency>

    <!-- hbase client 1.2.6 -->
    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-client</artifactId>
        <version>1.2.6</version>
    </dependency>
</dependencies>

spring boot 配置habse 地址

在application.yml中：

hbase:
  zookeeper:
    quorum: node202.hmbank.com,node203.hmbank.com,node204.hmbank.com
    port: 2181

HbaseTemplate bean定义

在spring boot 的 Application类中 使用@Bean注解加载：

@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        // TODO Auto-generated method stub
        
        SpringApplication.run(Application.class, args);
    }
     @Bean
     public HbaseTemplate hbaseTemplate(@Value("${hbase.zookeeper.quorum}") String quorum,
                                           @Value("${hbase.zookeeper.port}") String port) {
            HbaseTemplate hbaseTemplate = new HbaseTemplate();
            org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
            conf.set("hbase.zookeeper.quorum", quorum);
            conf.set("hbase.zookeeper.port", port);
            hbaseTemplate.setConfiguration(conf);
            hbaseTemplate.setAutoFlush(true);
            
           // hbaseTemplate.get("test1", "0001", null);
            
            return hbaseTemplate;
        }

hbase dao的设计使用

1. 定义dao 类
2. 使用@Component注解，使该dao可以被spring boot加载
3. 注入 HbaseTemplate

@Component
public class ClientTableDao {

    private Logger logger = LoggerFactory.getLogger(this.getClass());
    
    @Autowired
    private HbaseTemplate hbaseTemplate;

4. 定义查询方法， 注意RowMapper 的定义
   示例如下：

    private void queryRelationships(String idno, String name, RShips rships) {
        String id = idno + "|" + name;
        this.hbaseTemplate.get(CLIENT_TABLE, id, RELATION_FAMILY , new RowMapper<RShips>() {

            @Override
            public RShips mapRow(Result result, int rowNum) throws Exception {
                List<Cell> ceList =   result.listCells();
                
                if (ceList == null || ceList.size() == 0) {
                    logger.debug("查询为空，返回！" + idno + name);
                    return null;
                }
                logger.debug("chaxundao de " + ceList.size());
                if(ceList!=null && ceList.size()>0){
                    for(Cell cell:ceList){
                        String value = Bytes.toString( cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                        String qualifier = Bytes.toString( cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                        
                        int ivalue = Integer.valueOf(value);
                        
                        rships.createShip(id, idno, name, qualifier, ivalue);
                    }
                }
                return null;
            }
        });
    }

        list = this.hbaseTemplate.get(CLIENT_TABLE, id, LABELS_FAMILY , new RowMapper<List<ClientLableData>>() {

            @Override
            public List<ClientLableData> mapRow(Result result, int rowNum) throws Exception {
                List<Cell> ceList =   result.listCells();
                List<ClientLableData> rrlist = new ArrayList<ClientLableData>();
                
                if(ceList!=null&&ceList.size()>0){
                    for(Cell cell:ceList){
                        //String value = Bytes.toString( cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                        //f = Bytes.toString( cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
                        String qualifier = Bytes.toString( cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                        
                        //double dvalue = Double.valueOf(value) Bytes.tod;
                        double dvalue = Bytes.toDouble(cell.getValueArray(), cell.getValueOffset() ); // , cell.getValueLength());
                        rrlist.add(new ClientLableData(idno, name, qualifier, dvalue));
                    }
                }
                return rrlist;
            }
        });

插入

    public void insertLabels(String id, String label, double value) {
        this.hbaseTemplate.put(CLIENT_TABLE, id, LABELS_FAMILY, label, Bytes.toBytes(value));;
    }