SpringBoot操作hbase

港控/mmm° 2023-09-30 14:11 50阅读 0赞

1，添加依赖（客户端版本和 HBase 版本需要保持一致，否则可能会遇到不兼容的问题。）

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
                <exclusions>
                    <exclusion>
                        <groupId>org.junit.vintage</groupId>
                        <artifactId>junit-vintage-engine</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.apache.hbase</groupId>
                <artifactId>hbase-shaded-client</artifactId>
                <version>1.2.0</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
            </dependency>
        </dependencies>

2，yml配置

spring:
      application:
        name: sbhbase
    server:
      port: 9090
    hbase:
      zookeeper:
        quorum: 192.168.88.180:2181

3，代码段（配置类，获取配置）

package com.qjc.sbhbase.config;
     
    import org.apache.hadoop.hbase.HBaseConfiguration;
    import org.apache.hadoop.hbase.HConstants;
    import org.apache.hadoop.hbase.client.Connection;
    import org.apache.hadoop.hbase.client.ConnectionFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Scope;
     
    import java.io.IOException;
    import java.util.function.Supplier;
     
    @Configuration
    public class CustomerHbaseConfiguration {
        
        @Value("${
          hbase.zookeeper.quorum}")
        private String quorum;
     
        @Bean
        public org.apache.hadoop.conf.Configuration getConfig(){
        
            org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
            conf.set(HConstants.ZOOKEEPER_QUORUM,quorum);
            return conf;
        }
        //每次用户调用get方法获得一个新的数据库连接  可以考虑并发
        @Bean
        public Supplier<Connection> hbaseConnSupplier(){
        
            return ()->{
        
                return hbaseConnect();
            };
        }
        //获取数据库连接
        @Bean
        @Scope("prototype")
        public Connection hbaseConnect(){
        
            Connection connection = null;
            try {
        
                connection = ConnectionFactory.createConnection(getConfig());
            } catch (IOException e) {
        
                e.printStackTrace();
            }
            return connection;
        }
    }

4,实体类

package com.qjc.sbhbase.model;
     
    import lombok.AllArgsConstructor;
    import lombok.Builder;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Scope;
     
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    @Builder
    @Scope("prototype")
    public class Userinfos {
        
        private String userid;
        private String username;
        private String birthday;
    }

5，操作（service）

package com.qjc.sbhbase.services;
     
    import com.qjc.sbhbase.model.Userinfos;
    import org.apache.hadoop.hbase.TableName;
    import org.apache.hadoop.hbase.client.*;
    import org.apache.hadoop.hbase.filter.CompareFilter;
    import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
    import org.apache.hadoop.hbase.filter.SubstringComparator;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
     
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
     
    @Service
    public class HbaseService {
        
        @Autowired
        private Connection hbaseConnection;
     
        //插入数据
        public void insert(Userinfos user){
        
            try {
        
                //获取数据库中的表
                Table table = null;
                table = hbaseConnection.getTable(TableName.valueOf("mydemo:userinfos"));
                //准备一行数据
                Put line = new Put(user.getUserid().getBytes());
                line.addColumn("base".getBytes(),"username".getBytes(),user.getUsername().getBytes());
                line.addColumn("base".getBytes(),"birthday".getBytes(),user.getBirthday().getBytes());
                //将数据插入数据库
                table.put(line);
            } catch (IOException e) {
        
                e.printStackTrace();
            }
        }
     
        //根据rowkey获取数据
        public Userinfos findByRowkey(String rowkey){
        
            Table table = null;
            Result result = null;
            Userinfos us = null;
            //获取hbase中的表
            try {
        
                table = hbaseConnection.getTable(TableName.valueOf("mydemo:userinfos"));
                //按照rowkey获取数据
                Get get = new Get(rowkey.getBytes());
                result = table.get(get);
                us = Userinfos.builder().username(new String(result.getValue("base".getBytes(),"username".getBytes()))).build();
            } catch (IOException e) {
        
                e.printStackTrace();
            }finally {
        
                try {
        
                    table.close();
                } catch (IOException e) {
        
                    e.printStackTrace();
                }
            }
            return us;
        }
     
     
        //代码逻辑没问题  在海量数据下此方法不能使用 内存不足 会花费大量时间下载数据
        public List<Userinfos> findAll(){
        
            List<Userinfos> list = new ArrayList<Userinfos>();
            Table table = null;
            try {
        
                table = hbaseConnection.getTable(TableName.valueOf("mydemo:userinfos"));
                Scan scan = new Scan();
                ResultScanner rs = table.getScanner(scan);
                Result result = null;
                while ((result =rs.next())!= null){
        
                    list.add(Userinfos.builder().username(new String(result.getValue("base".getBytes(),"username".getBytes()))).build());
                }
            } catch (IOException e) {
        
                e.printStackTrace();
            }finally {
        
                try {
        
                    table.close();
                } catch (IOException e) {
        
                    e.printStackTrace();
                }
            }
            return list;
        }
     
        //根据username包含某字查询信息
        public List<Userinfos> findUserByPrefixName(String prefixName){
        
            Table table = null;
            List<Userinfos> list = new ArrayList<>();
            try {
        
                table = hbaseConnection.getTable(TableName.valueOf("mydemo:userinfos"));
                Scan scan = new Scan();
                //hbase中操纵命令：scan 'mydemo:userinfos',{FILTER=>"SingleColumnValueFilter('base','username',=,'substring:zhang')"}
                SingleColumnValueFilter vf = new SingleColumnValueFilter(
                        "base".getBytes(),
                        "username".getBytes(),
                        CompareFilter.CompareOp.EQUAL,
                        new SubstringComparator(prefixName)
                );
                scan.setFilter(vf);
                ResultScanner scanner = table.getScanner(scan);
     
                Result rs = null;
                while ((rs=scanner.next()) != null){
        
                    list.add(Userinfos.builder().username(new String(rs.getValue("base".getBytes(),"username".getBytes()))).build());
                }
            } catch (IOException e) {
        
                e.printStackTrace();
            } finally {
        
                try {
        
                    table.close();
                } catch (IOException e) {
        
                    e.printStackTrace();
                }
            }
            return list;
        }
    }

6，controller层（这里只是简单调用查看）

package com.qjc.sbhbase.controller;
     
    import com.qjc.sbhbase.model.Userinfos;
    import com.qjc.sbhbase.services.HbaseService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.ModelAttribute;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
     
    import javax.annotation.Resource;
    import java.util.List;
     
    @RestController
    public class InitCtrl {
        
        @Resource
        private HbaseService hbaseService;
     
        @RequestMapping("/add")
        public String add(@ModelAttribute Userinfos user){
        
            hbaseService.insert(user);
            return "ok";
        }
     
        @RequestMapping("/single/{rowkey}")
        public Userinfos select(@PathVariable("rowkey") String rowkey){
        
            return hbaseService.findByRowkey(rowkey);
        }
     
        @RequestMapping("/getAll")
        public List<Userinfos> getAll(){
        
            return hbaseService.findAll();
        }
     
        @RequestMapping("/findByName/{name}")
        public List<Userinfos> findByName(@PathVariable("name") String name){
        
            return hbaseService.findUserByPrefixName(name);
        }
    }