Hbase核心API操作

痛定思痛。 2023-01-08 10:22 51阅读 0赞

# Hbase核心API操作 #

## 一、获取Configuration对象 ##

> Connection代表对集群的连接对象，封装了与实际服务器的低级别单独连接以及与zookeeper的连接。  
> Connection可以通过ConnectionFactory类实例化。  
> Connection的生命周期由调用者管理，使用完毕后需要执行close()以释放资源。  
> Connection是线程安全的，多个Table和Admin可以共用同一个Connection对象。因此一个客户端只需要实例化一个连接即可。  
> 反之，Table和Admin不是线程安全的！因此不建议并缓存或池化这两种对象。

package com.hbase.use;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.hbase.HBaseConfiguration;
    import org.apache.hadoop.hbase.client.Connection;
    import org.apache.hadoop.hbase.client.ConnectionFactory;
    
    import java.io.IOException;
    
    /** * 创建对象的方法： * 1.构造器 * 2.工厂模式 XXXFactory.createXXX * 3.建造者模式 XXXBuilder.build() * * Connection:代表和集群的一个连接，会维护一个和hbaseserver的连接以及和zookeeper的连接 * 通过ConnectionFactory获取一个连接的实例 * 由调用者维护生命周期，调用close关闭连接释放资源！ * Connection不仅负责连接集群，还负责维护meta的缓存，负责定位region等，这些都可以被连接产生的Table和Admin对象共享！ * Connection的创建是重量级的，不建议频繁创建！是线程安全的，因此建议在一个应用中只创建一次连接，在多个线程中可以共享这一个connection对象！ * * Table和Admin对象是轻量级的，不是线程安全的，因此建议每个线程都拥有自己独立的Table和Admin对象。不建议池化或者缓存！ * * DDL: Admin * DML：Table */
    
    public class ConnectionUtil { 
    	//创建连接对象
        public static Connection connection() throws IOException { 
    		//如何创建Hadoop的Configuration , new对象，默认加载 8个 xxx-default.xml和xxx-site.xml的配置文件
            // 除了hadoop的配置文件，还会读取hbase的配置文件
            Configuration conf = HBaseConfiguration.create();
    
            Connection connection = ConnectionFactory.createConnection(conf);
    
            return connection;
        }
    
    	//创建关闭连接对象
        public static void closeConn(Connection conn) throws IOException { 
    
            if (conn != null){ 
                conn.close();
            }
        }
    }

## 二、库的相关操作API ##

package com.hbase.use;
    
    import org.apache.commons.lang.StringUtils;
    import org.apache.hadoop.hbase.NamespaceDescriptor;
    import org.apache.hadoop.hbase.client.Admin;
    import org.apache.hadoop.hbase.client.Connection;
    import org.apache.hadoop.hbase.client.TableDescriptor;
    import org.apache.hadoop.hbase.util.Bytes;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    
    public class NamespaceTest { 
    
        public static Logger logger = LoggerFactory.getLogger(NamespaceTest.class);
    
        //查看所有命名空间
        public static List<String>  listNamespace(Connection conn) throws IOException { 
    
            List<String> nss = new ArrayList<>();
    
            //获取Admin对象
            Admin admin = conn.getAdmin();
    
            NamespaceDescriptor[] namespaceDescriptors = admin.listNamespaceDescriptors();
    
            for (NamespaceDescriptor namespaceDescriptor : namespaceDescriptors) { 
    
                nss.add(namespaceDescriptor.getName());
    
            }
    
            //关闭Admin
            admin.close();
    
            return nss;
        }
    
    
    
        //判断库是否存在
        public static boolean  ifNamespaceExits(Connection conn,String nsName) throws IOException { 
    
            //判断入参
            if (StringUtils.isBlank(nsName)){ 
                logger.error("库名非法");
                return false;
            }
    
            //获取Admin对象
            Admin admin = conn.getAdmin();
    
            try { 
                admin.getNamespaceDescriptor(nsName);
                return true;
            } catch (IOException e) { 
                //e.printStackTrace();
                return false;
            }finally { 
                //关闭Admin
                admin.close();
            }
        }
    
    
        //创建库
        public static boolean  createNamespace(Connection conn,String nsName) throws IOException { 
    
            //判断入参
            if (StringUtils.isBlank(nsName)){ 
                logger.error("输入参数有误");
                return false;
            }
    
            //获取Admin对象
            Admin admin = conn.getAdmin();
            NamespaceDescriptor build = NamespaceDescriptor.create(nsName).build();
    
            try { 
                admin.createNamespace(build);
                return true;
            } catch (IOException e) { 
                //e.printStackTrace();
                return false;
            }finally { 
                //关闭Admin
                admin.close();
            }
        }
    
    
    
        //删除库
        public static boolean  deleteNamespace(Connection conn,String nsName) throws IOException { 
    
            //判断入参
            if (StringUtils.isBlank(nsName)){ 
                logger.error("输入参数有误");
                return false;
            }
    
            //判断是否是空库，只能删除空库
            List<String> strings = listTableInNamespace(conn, nsName);
    
            if (strings==null || strings.size()>0){ 
                logger.error("库中有表，或者库不存在！");
                return false;
            }
    
            //获取Admin对象
            Admin admin = conn.getAdmin();
    
            try { 
                admin.deleteNamespace(nsName);
                return true;
            } catch (IOException e) { 
                e.printStackTrace();
                return false;
            }finally { 
                //关闭Admin
                admin.close();
            }
    
        }
    
    
    
        //获取库中所有表
        public static List<String>  listTableInNamespace(Connection conn,String nsName) throws IOException { 
    
            //判断入参
            if (StringUtils.isBlank(nsName)){ 
                logger.error("库名非法");
                return null;
            }
    
            //判断库是否存在
            if (!ifNamespaceExits(conn,nsName)){ 
                logger.error("库不存在");
                return null;
            }
    
            List<String> tables = new ArrayList<>();
    
            //获取Admin对象
            Admin admin = conn.getAdmin();
    
            List<TableDescriptor> tableDescriptors = admin.listTableDescriptorsByNamespace(Bytes.toBytes(nsName));
    
            for (TableDescriptor tableDescriptor : tableDescriptors) { 
                tables.add(tableDescriptor.getTableName().getNameAsString());
            }
    
            //关闭Admin
            admin.close();
    
            return tables;
    
        }
    
    }

## 三、表的相关操作 ##

package com.hbase.use;
    
    import javafx.scene.control.TableColumn;
    import org.apache.commons.lang.StringUtils;
    import org.apache.hadoop.hbase.TableName;
    import org.apache.hadoop.hbase.client.*;
    import org.apache.hadoop.hbase.util.Bytes;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    
    public class TableUtils { 
    
        public static Logger logger = LoggerFactory.getLogger(TableUtils.class);
    
        //判断表名是否合法
        public static TableName isTableNameLegal(Connection conn,String nsn, String tn) throws IOException { 
    
            //判断入参
            if (StringUtils.isBlank(tn)){ 
                logger.error("表名非法！");
                return null;
            }
    
            //如果未指定库名，返回default库
            if (nsn==null){ 
                return TableName.valueOf(nsn,tn);
            }else if(NamespaceTest.ifNamespaceExits(conn,nsn)){ 
                return TableName.valueOf(nsn,tn);
            }else { 
                logger.error("库非法！");
                return null;
            }
        }
    
    
    
        //判断表是否存在
        public static boolean isTableExit(Connection conn,String nsn, String tn) throws IOException { 
    
            //判断表名是否非法
            TableName tableNameLegal = isTableNameLegal(conn, nsn, tn);
    
            //如果未指定库名，返回default库
            if (tableNameLegal==null){ 
                return false;
            }
    
            Admin admin = conn.getAdmin();
    
            boolean result = admin.tableExists(tableNameLegal);
    
            admin.close();
    
            return result;
        }
        
    
    
        //删除表
        public static boolean deleteTable(Connection conn,String nsn, String tn) throws IOException { 
    
            //判断表名是否非法
            TableName tableNameLegal = isTableNameLegal(conn, nsn, tn);
    
            //如果未指定库名，返回default库
            if (tableNameLegal==null){ 
                return false;
            }
    
            //获取admin对象
            Admin admin = conn.getAdmin();
    
            //先禁用表
            admin.disableTable(tableNameLegal);
            //再删除表
            admin.deleteTable(tableNameLegal);
            //关闭admin
            admin.close();
    
            return true;
    
        }
        
    
    
        //创建表
        public static boolean   createTable(Connection conn,String nsn, String tn,String... cfNames) throws IOException { 
    
            //判断表名是否非法
            TableName tableNameLegal = isTableNameLegal(conn, nsn, tn);
    
            //如果未指定库名，返回default库
            if (tableNameLegal==null){ 
                //throw new RuntimeException("表名非法！");
                return false;
            }
    
            //获取admin对象
            Admin admin = conn.getAdmin();
    
            //创建表所有列族的描述
            //ColumnFamilyDescriptor代表一个列族的描述，在这个对象中设置列族的各种属性
            List<ColumnFamilyDescriptor> cfds = new ArrayList<>();
    
            for (String cfName : cfNames) { 
    
                ColumnFamilyDescriptor build = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cfName))
                        .setMaxVersions(5)
                        .build();
    
                cfds.add(build);
            }
    
            //使用TableDescriptorBuilder调用各种setXXX方法，为表设置各种属性，之后调用builder（）获取TableDescriptor对象
            TableDescriptor build = TableDescriptorBuilder.newBuilder(tableNameLegal)
                    .setColumnFamilies(cfds)
                    .build();
    
            admin.createTable(build);
    
            //关闭admin
            admin.close();
    
            return true;
        }
    
    }

## 四、数据相关操作 ##

package com.hbase.use;
    
    import org.apache.hadoop.hbase.Cell;
    import org.apache.hadoop.hbase.CellUtil;
    import org.apache.hadoop.hbase.TableName;
    import org.apache.hadoop.hbase.client.*;
    import org.apache.hadoop.hbase.util.Bytes;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.io.IOException;
    
    public class DataUtil { 
    
        public static Logger logger = LoggerFactory.getLogger(NamespaceTest.class);
    
        //获取table
        public static Table getTable(Connection conn,String nsn,String tn) throws IOException { 
    
            TableName tableName = TableUtils.isTableNameLegal(conn, nsn, tn);
    
            if (tableName == null){ 
                logger.error("输入表名有误");
                return null;
            }
            return conn.getTable(tableName);
        }
    
    
    
        //put 添加 | 修改表 内容
        public static void put(Connection conn,String nsn,String tm,String rowkey,String columnFamily,String columnName,String value) throws IOException { 
    
            Table table = getTable(conn, nsn, tm);
    
            if (table==null){ 
                logger.error("表不存在！");
                return;
            }
    
            //指定向哪一行插入或者修改
            //指定向当前行的那一列插入或者修改
            Put put = new Put(Bytes.toBytes(rowkey))
                    .addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName), Bytes.toBytes(value));
    
            //执行put操作
            table.put(put);
            //关闭资源
            table.close();
        }
    
    
    
        //get查询单行 get '表名','rowkey'
        //Result代表单行查询结果，一个result中含有多个Cell
        public static void get(Connection conn,String nsn,String tn,String rowkey) throws IOException { 
            Table table = getTable(conn, nsn, tn);
    
            if (table==null){ 
                logger.error("表不存在！");
                return;
            }
    
            //指定查询哪一行
            Get get = new Get(Bytes.toBytes(rowkey));
    
            Result result = table.get(get);
    
            if (result == null){ 
                logger.error("所查询行不存在！");
                return;
            }
    
            //Result代表单行的结果，是一行所有cell的结果
            Cell[] cells = result.rawCells();
    
            for (Cell cell : cells) { 
    
                String rk = Bytes.toString(CellUtil.cloneRow(cell));
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String cq = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
    
                System.out.println("rowkey: "+rk+" 列族： "+cf+" 列： "+cq+" 值： "+value);
            }
        }
        
    
    
        //scan scan "表名" {STARTRAW=>XXX,RAW=>TRUE,VERSIONS=>3}
        public static void scan(Connection conn,String nsn,String tn,String rawkey,String startRow,String stopRow) throws IOException { 
    
            Table table = getTable(conn, nsn, tn);
    
            if (table==null){ 
                logger.error("表不存在！");
                return;
            }
            
            Scan scan = new Scan().withStartRow(Bytes.toBytes(startRow)).withStopRow(Bytes.toBytes(stopRow));
    
            //ResultScanner是一个迭代器，可以迭代scan扫描多行的Result
            ResultScanner scanner = table.getScanner(scan);
    
            for (Result result : scanner) { 
    
                if (result == null) { 
                    logger.error("所查询行不存在！");
                    return;
                }
    
                //Result代表单行的结果，是一行所有cell的结果
                Cell[] cells = result.rawCells();
    
                for (Cell cell : cells) { 
                    String rk = Bytes.toString(CellUtil.cloneRow(cell));
                    String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                    String cq = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
    
                    System.out.println("rowkey: " + rk + " 列族： " + cf + " 列： " + cq + " 值： " + value);
                }
            }
            table.close();
    
        }
    
    
        public static void delete(Connection conn,String nsn,String tn,String rowkey,String cf,String cq) throws IOException { 
    
            Table table = getTable(conn, nsn, tn);
    
            if (table==null){ 
                logger.error("表不存在！");
                return;
            }
    
            //删除整行，hbase为这行所有列族，都添加一行：列族名：（空）,type=DeleteFamily,时间戳是最新
            Delete delete = new Delete(Bytes.toBytes(rowkey));
    
            //只删除某个列族，hbase为这个列族，都添加一行：列族名：（空）,type=DeleteFamily,时间戳是最新
            //delete.addFamily(Bytes.toBytes(cf));
    
            //只删除某个列，最新版本的数据，返回删除后的最近版本
            //hbase为这个列，添加了一个cell：列族名：列名，ts=最新版本数据的ts，type=delete
            //delete.addColumn(Bytes.toBytes(cf),Bytes.toBytes(cq));
    
            //删除某个列，所有版本的数据，为这个列，添加一个cell：列族名：列名，ts=最新ts，type=DeleteColumn
            delete.addColumns(Bytes.toBytes(cf),Bytes.toBytes(cq));
    
            table.delete(delete);
    
            table.close();
        }
    
        //遍历result
        public static void getResult(Result result) { 
    
            if (result == null) { 
                logger.error("所查询行不存在！");
                return;
            }
            //Result代表单行的结果，是一行所有cell的结果
            Cell[] cells = result.rawCells();
    
            for (Cell cell : cells) { 
    
                String rk = Bytes.toString(CellUtil.cloneRow(cell));
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String cq = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
    
                System.out.println("rowkey: " + rk + " 列族： " + cf + " 列： " + cq + " 值： " + value);
            }
        }
    }