JDBC操作HBase

什么是JDBC? JDBC 规范定义接口，具体的实现由各大数据库厂商来实现。 JDBC 是 Java 访问数据库的标准规范，真正怎么操作数据库还需要具体的实现类，也就是数据库驱动。每个数据库厂商根据自家数据库的通

NightFall丶

2140人浏览 · 2021-04-04 15:15:53

NightFall丶 · 2021-04-04 15:15:53 发布

什么是JDBC?

JDBC 规范定义接口，具体的实现由各大数据库厂商来实现。

JDBC 是 Java 访问数据库的标准规范，真正怎么操作数据库还需要具体的实现类，也就是数据库驱动。每个数据库厂商根据自家数据库的通信格式编写好自己数据库的驱动。所以我们只需要会调用 JDBC 接口中的方法即可，数据库驱动由数据库厂商提供。

核心API

初始化连接与mysql/hive不同的时，无需加载驱动，通过ConnectionFactory直接可以创建Connection;

Connection 接口：数据级操作使用对象(e.g. 增删改查数据)

Admin接口：对象级操作对象(e.g. 创建Namespace，创建表…)

Code

Linking Denpency

 <dependency>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-client</artifactId>
      <version>1.2.0-cdh5.14.2</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-common</artifactId>
      <version>1.2.0-cdh5.14.2</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-server</artifactId>
      <version>1.2.0-cdh5.14.2</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-common</artifactId>
      <version>2.6.0-cdh5.14.2</version>
    </dependency>
    <dependency>
      <groupId>log4j</groupId>
      <artifactId>log4j</artifactId>
      <version>1.2.17</version>
    </dependency>

初始化类

package cn.wsj;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

import java.io.IOException;

public class DbUtils {
    //单例对象
    private DbUtils(){};
    private static Connection con;

    /**
     * 初始化参数配置
     * 创建Connection
     * 调用时先判断是否存在，存在的直接返回Connection对象
     * @return
     */
    public static Connection getCon(){
        try {
            if(con==null){
                Configuration config = new Configuration();
                config.set("hbase.zookeeper.quorum","sole");
                config.set("hbase.zookeeper.properties.clientPort","2181");
                config.set("hbase.master","sole:16000");
                con = ConnectionFactory.createConnection(config);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return con;
    }
}

主要方法类

package cn.wsj;

import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class BaseDao {
    private Connection  con;
    private Admin admin;

    /**
     * 初始化Connection和Admin实例
     * 对象级操作->Admin
     * 数据级操作->Connection
     * @throws IOException
     */
    public BaseDao() throws IOException {
        con = DbUtils.getCon();
        admin = con.getAdmin();
    }

    /**
     * 将所有namespace统一存入list，方便调用
     * @return
     * @throws IOException
     */
    private List<String> nameSpaces() throws IOException {
        List<String> names = new ArrayList<>();
        NamespaceDescriptor[] namespaces = admin.listNamespaceDescriptors();
        for (NamespaceDescriptor namespace : namespaces) {
            names.add(namespace.getName());
        }
        return names;
    }

    /**
     * 查看所有namespace
     * @throws IOException
     */
    public void listNamespace() throws IOException {
        List<String> lst = nameSpaces();
        lst.forEach(x-> System.out.println(x));
    }

    /**
     * 新增namespace
     * 先判断是否存在，不存在则创建
     * @throws IOException
     */
    public void addNamespace(String namespace) throws IOException {
        List<String> lst = nameSpaces();
        if(!lst.contains(namespace)){
            admin.createNamespace(NamespaceDescriptor.create(namespace).build());
        }else{
            System.err.println("Namespace: "+namespace+" has exists!");
            return;
        }
    }

    /**
     * 将此命名空间下的表统一存放，方便调用
     * @param namespace
     * @return
     * @throws IOException
     */
    private List<String> listTables(String namespace) throws IOException {
        List<String> tables = new ArrayList<>();
        HTableDescriptor[] HTables = admin.listTableDescriptorsByNamespace(namespace);
        for (HTableDescriptor HTable : HTables) {
            tables.add(HTable.getNameAsString());
        }
        return tables;
    }

    /**
     * 删除命名空间前，判断命名空间是否存在
     * 存在的前提下要先删除所有表才可删除命名空间
     * @param namespace
     * @throws IOException
     */
    public void deleteNamespace(String namespace) throws IOException {
        List<String> names = nameSpaces();
        if(names.contains(namespace)){
            List<String> lst = listTables(namespace);
            for (String table : lst) {
                TableName tb = TableName.valueOf(table);
                admin.disableTable(tb);
                admin.deleteTable(tb);
            }
            admin.deleteNamespace(namespace);
        }else{
            System.err.println("Namespace: "+namespace+" not exists!");
        }
    }

    /**
     * 通过namespace获取其所有的表
     * @param namespace
     * @throws IOException
     */
    public void showTables(String namespace) throws IOException {
        List<String> lst = listTables(namespace);
        lst.forEach(x-> System.out.println(x));
    }

    /**
     * 先判断表是否存在
     * 不存在才可创建
     * @param tableName
     * 表名组成：Namespace:TableName
     * @throws IOException
     */
    public void createTable(String tableName) throws IOException {
        String[] split = tableName.split(":");
        List<String> lst = listTables(split[0]);
        if(!lst.contains(tableName)) {
            TableName tb = TableName.valueOf(tableName);
            HTableDescriptor descriptor = new HTableDescriptor(tb);
            descriptor.addFamily(new HColumnDescriptor("info"));
            descriptor.addFamily(new HColumnDescriptor("score"));
            admin.createTable(descriptor);
        }else {
            System.err.println("Table: "+tableName+" has exists!");
        }
    }

    /**
     * 表存在则删除，不存在则退出
     * @param tableName
     * 表名组成：Namespace:TableName
     * @throws IOException
     */
    public void deleteTable(String tableName) throws IOException {
        String[] split = tableName.split(":");
        List<String> lst = listTables(split[0]);
        if (lst.contains(tableName)) {
            TableName tb = TableName.valueOf(tableName);
            admin.disableTable(tb);
            admin.deleteTable(tb);
        }else {
            System.err.println("Table: "+tableName+" not exists!");
        }
    }

    /**
     * 将字符串转为字节数组，编码格式为UTF_8
     * @param s
     * @return
     */
    private byte[] to(String s){
        return s.getBytes(StandardCharsets.UTF_8);
    }

    /**
     * 将字节数组转为字符串，编码格式为UTF_8
     * @param b
     * @return
     */
    private String to(byte[] b){
        return new String(b, StandardCharsets.UTF_8);
    }

    /**
     * 将map层层拆解，之后组成List<Put>返回
     * @param data
     * @return
     */
    private List<Put> map(Map<String,Map<String, Map<String,Object>>> data){
        List<Put> lst = new ArrayList<>(data.size());
        for (Map.Entry<String, Map<String, Map<String, Object>>> rowkeySet : data.entrySet()) {
            String rowkey = rowkeySet.getKey();
            Put put = new Put(to(rowkey));
            for (Map.Entry<String, Map<String, Object>> familySet : rowkeySet.getValue().entrySet()) {
                String family = familySet.getKey();
                for (Map.Entry<String, Object> colVal : familySet.getValue().entrySet()) {
                    String column = colVal.getKey();
                    Object value = colVal.getValue();
                    put.addColumn(to(family),to(column),to(value.toString()));
                }
                lst.add(put);
            }
        }
        return lst;
    }

    /**
     *  插入数据需要Put对象
     * @param tableName
     * @param data
     * data最外层为rowkey,向内一层为family,最内一层为column和value
     */
    public void insertData(String tableName,Map<String,Map<String, Map<String,Object>>> data) throws IOException {
        TableName tb = TableName.valueOf(tableName);
        Table table = con.getTable(tb);
        table.put(map(data));
        table.close();
    }

    /**
     * 将CellScanner不断读取信息并打印
     * @param cs
     * @throws IOException
     */
    private void print(CellScanner cs) throws IOException {
        while(cs.advance()){
            Cell cell = cs.current();
            String row = to(CellUtil.cloneRow(cell));
            String family = to(CellUtil.cloneFamily(cell));
            String column = to(CellUtil.cloneQualifier(cell));
            String value = to(CellUtil.cloneValue(cell));
            String s = MessageFormat.format("{0}\t{1}\t{2}\t{3}", row, family, column, value);
            System.out.println(s);
        }
    }


    /**
     * 查看table对应rowkey下的所有信息
     * @param tableName
     * @throws IOException
     */
    public void showData(String tableName,String rowkey) throws IOException {
        TableName tb = TableName.valueOf(tableName);
        Table table = con.getTable(tb);
        Get get = new Get(to(rowkey));
        Result rst = table.get(get);
        CellScanner scanner = rst.cellScanner();
        print(scanner);
    }


    /**
     * 查看table对应rowkey对应列族下的所有信息
     * @param tableName
     * @param rowkey
     * @param family
     * @throws IOException
     */
    public void showData(String tableName,String rowkey,String family) throws IOException {
        TableName tb = TableName.valueOf(tableName);
        Table table = con.getTable(tb);
        Get get = new Get(to(rowkey));
        get.addFamily(to(family));
        Result rst = table.get(get);
        CellScanner scanner = rst.cellScanner();
        print(scanner);
    }

    /**
     * 查看table对应rowkey对应列族对应列的信息
     * @param tableName
     * @param rowkey
     * @param family
     * @param column
     * @throws IOException
     */
    public void showData(String tableName,String rowkey,String family,String column) throws IOException {
        TableName tb = TableName.valueOf(tableName);
        Table table = con.getTable(tb);
        Get get = new Get(to(rowkey));
        get.addColumn(to(family),to(column));
        Result rst = table.get(get);
        CellScanner scanner = rst.cellScanner();
        print(scanner);
    }

    /**
     * 删除表中对应rowkey的所有数据
     * @param tableName
     * @param rowkey
     * @throws IOException
     */
    public void deleteDate(String tableName,String rowkey) throws IOException {
        TableName tb = TableName.valueOf(tableName);
        Table table = con.getTable(tb);
        Delete delete = new Delete(to(rowkey));
        table.delete(delete);
    }

    /**
     * 删除表中对应rowkey对应列族下的所有数据
     * @param tableName
     * @param rowkey
     * @param family
     * @throws IOException
     */
    public void deleteDate(String tableName,String rowkey,String family) throws IOException {
        TableName tb = TableName.valueOf(tableName);
        Table table = con.getTable(tb);
        Delete delete = new Delete(to(rowkey));
        delete.addFamily(to(family));
        table.delete(delete);
    }


    /**
     * 删除表中对应rowkey对应列族对应列的数据
     * @param tableName
     * @param rowkey
     * @param family
     * @param column
     * @throws IOException
     */
    public void deleteDate(String tableName,String rowkey,String family,String column) throws IOException {
        TableName tb = TableName.valueOf(tableName);
        Table table = con.getTable(tb);
        Delete delete = new Delete(to(rowkey));
        delete.addColumn(to(family),to(column));
        table.delete(delete);
    }

    /**
     * 自动关闭
     * @param closeables
     */
    private void close(AutoCloseable...closeables){
        for (AutoCloseable closeable : closeables) {
            if(null!= closeable){
                try {
                    closeable.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }


    public void shut(){
        close(admin,con);
    }

}