Hbase高手之路 – 第五章 – HBase的Java API编程

一、 需求与数据集

某自来水公司，需要存储大量的缴费明细数据，以下截取了缴费明细的一部分内容：

因为缴费明细的数据记录非常庞大，该公司的信息部门决定使用hbase来存储这些数据，并且可以使用java来访问这些数据。

二、 准备工作

idea2021.3.3 专业版 (社区版也可以)

1. 下载安装idea

2. 配置国内的maven镜像库

3. 创建一个maven工程

4. 修改pom文件，导入相关的依赖

5. 复制HBase的配置文件hbase-site.xml到resources目录中

先导出到本地计算机

6. 创建包结构

7. 创建hbase连接类及管理对象

public class HbaseConnect {
    public static void main(String[] args){
        //1、创建hbase的配置
        Configuration configuration = HBaseConfiguration.create();
        //2、创建hbase的连接
        Connection connection;

        {
            try {
                connection = ConnectionFactory.createConnection(configuration);
                System.out.println(connection);
                //3、创建admin对象
                Admin admin = connection.getAdmin();
                System.out.println(admin);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}

三、 案例一、使用java api创建hbase的表

创建一个名为water_bill的表，包含一个列簇info。

1. 编写代码

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class HbaseJavaAPIDemo2 {
    public static void main(String[] args) throws IOException {
        Configuration configuration = HBaseConfiguration.create();
        // 2.	使用ConnectionFactory.createConnection()创建Hbase连接
        Connection connection = ConnectionFactory.createConnection(configuration);
        Admin admin = null;
        {
            try {
                //通过工厂模式，根据配置来创建连接
                connection = ConnectionFactory.createConnection(configuration);
                System.out.println(connection);
                //3.创建admin对象
                admin = connection.getAdmin();
                System.out.println(admin);
                //4.创建water_bill的表，包含一个列簇info
                //定义表名
                TableName tableName = TableName.valueOf("WATER_BILL3");
                // 1) 构建表描述构建器构建器模式
                TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
                // 2)构建列簇描述构建器
                ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info"));
                // 3)定义列簇描述对象
                ColumnFamilyDescriptor columnFamilyDescriptor = columnFamilyDescriptorBuilder.build();
                // 4)表描述构建器和列簇描述对象建立关系
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
                // 5)定义表描述对象
                TableDescriptor tableDescriptor = tableDescriptorBuilder.build();
                // 6)创建表
                admin.createTable(tableDescriptor);
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                assert admin != null;
                admin.close();
                connection.close();
            }
        }
    }
}

2. 运行

3. 查看结果

四、 案例二：使用java api删除表

删除刚刚创建的表

1. 编写代码

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;


public class HbaseJavaAPIMain {
    public static void main(String[] args) throws IOException {
        Admin admin = getAdmin();
        //调用删除表的方法
        boolean result = deleteTable(admin,"WATER_BILL");
        if(result) {
            System.out.println("删除成功");
        }else {
            System.out.println("表不存在");
        }
        //调用插入数据的方法
//        putTable(admin.getConnection(),"WATER_BILL3","4944191","info","name","张三");
//        关闭admin
        admin.close();
    }
    public static Admin getAdmin() {
        //1.创建hbase配置
        Configuration configuration = HBaseConfiguration.create();
        // 2.	使用ConnectionFactory.createConnection()创建Hbase连接
        Connection connection;
        Admin admin =null;
        try {
            //通过工厂模式，根据配置来创建连接
            connection = ConnectionFactory.createConnection(configuration);
            //创建admin对象
            admin = connection.getAdmin();
        }catch (IOException e){
            e.printStackTrace();
        }
        return admin;
    }

    /**
     * 向表中插入数据
     * @param connection
     * @param tableName
     * @param haha
     * @param columnFamily
     * @param column
     * @param value
     * @throws IOException
     */
    public static void putTable(Connection connection,String tableName,String haha,String columnFamily,String column,String value) throws IOException {
        //获取table对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        //根据 rowkey 获取Put对象
        Put put = new Put(Bytes.toBytes(haha));
        //添加姓名列
        put.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(column),Bytes.toBytes(value));
        //插入数据
        table.put(put);
        //关闭table
        table.close();

    }

    /**
     * 删除表
     * @param admin
     * @param name
     * @return
     * @throws IOException
     */
    public static boolean deleteTable(Admin admin,String name)throws IOException {
        //定义表名
        TableName tableName = TableName.valueOf(name);
        //判断表是否存在
        if (admin.tableExists(tableName)) {
            //禁用表
            admin.disableTable(tableName);
            //删除表
            admin.deleteTable(tableName);
            return  true;
        } else {
            return false;
        }

    }
}

2. 查看结果

五、 案例三：向创建的表中插入数据

向WATER_BILL中插入姓名列的数据

1. 编写代码

2. 调用方法

3. 查看结果

六、 案例：插入其他列的数据

插入地址列的数据

1. 编写代码

2. 调用方法

3. 查看结果

七、 案例四：查看一条数据

查询显示rowkey为4944191的所有列的数据

1. 编写方法

(1) 获取某列的值

(2) 获取某行的数据

2. 调用方法

八、 案例五：导入数据

1. 需求

有一份10w条记录的抄表数据文件，需要将其导入hbase中

2. import job导入大量数据

在hbase中，有一个import的MR作业，可以专门用来将数据导入到hbase中
用法：
hbase org.apache.hadoop.hbase.mapreduce.import 表名 hdfs 数据文件路径

3. 上传数据到hdfs

4. 导入数据

(1) 启动yarn

(2) 创建表

(3) 运行导入命令

hbase org.apache.hadoop.hbase.mapreduce.Import WATER_BILL /water_bill/

运行结果如下

(4) 查看数据

(5) count计数

(6) mapreduce计数

hbase org.apache.hadoop.hbase.mapreduce.RowCounter "WATER_BILL"

九、 案例六：查询2020年6月所有用户的用水量

1. 需求分析

在hbase中用scan+filter实现过滤查询。2020年6月份其实就是从2020年6月1号到2020年6月30日的所有抄表数据

2. 编写代码

public static void queryTable(Connection connection,String tableName,String columnFamily, String column,String startValue,String endValue)throws IOException {
        //1.获取Table
        Table table = connection.getTable(TableName.valueOf(tableName));
        //2.构建scan对象
        Scan scan = new Scan();
        //3.1构建过滤器
        //构建日期范围的过滤器
//        构建开始日期的过滤器
        SingleColumnValueFilter startDateFilter = new SingleColumnValueFilter(Bytes.toBytes(columnFamily),Bytes.toBytes(column),
                CompareOperator.GREATER_OR_EQUAL,Bytes.toBytes(startValue));
        //构建结束日期的过滤器
        SingleColumnValueFilter endDateFilter = new SingleColumnValueFilter(Bytes.toBytes(columnFamily),Bytes.toBytes(column),
                CompareOperator.LESS_OR_EQUAL,Bytes.toBytes(endValue));
        //3.2构建过滤器列表
        FilterList filterlist = new FilterList(FilterList.Operator.MUST_PASS_ALL,startDateFilter,endDateFilter);
        //4.构建扫描器
        scan.setFilter(filterlist);
        //5.执行scan扫描操作
        ResultScanner resultScanner = table.getScanner(scan);
        //6.迭代打印result
        for(Result result:resultScanner){
            // 6.1打印rowkey
            System.out.println("rowkey=>" + Bytes.toString(result.getRow()));
            System.out.println("-------------------------------------------");
            // 6.2构建单元格列表
            List<Cell> cells = result.listCells();
            for(Cell cell:cells){
                // 6.3 打印列簇名
                System.out.print(Bytes.toString(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength()));
                //打印列名
                String columnName = Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength());
                System.out.print(":" + columnName);
//                System.out.print(":" + Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()));
                //解决数值型乱码打印值
                //判断是否为数值型列
                if (columnName.equals("NUM_CURRENT") || columnName.equals("NUM_PREVIOUS")
                        || columnName.equals("NUM_USAGE") || columnName.equals("TOTAL_MONEY")) {
                    //打印数值型
                    System.out.println("=>" + Bytes.toDouble(cell.getValueArray(),cell.getValueOffset()));
                } else {
                    //打印字符串值
                    System.out.println("=>" + Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()));
                }
                //打印值
//                System.out.println("=>" + Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()));
            }
            System.out.println("-------------------------------------------");
        }
        //关闭资源
        resultScanner.close();
        table.close();
    }


}

3. 调用方法

4. 查看结果

5. 代码改进

改进后的结果

6. 解决数值型数据显示乱码的问题

打印显示字符串数据正常，但是如果hbase存储的是int、double、float等数值型数据时，显示就会乱码，解决的方法就是判断是否是数值型数据，如果是，则进行相应的转化。

(1) 代码改进

(2) 显示结果

乱码问题已经解决了

十、 案例七：Export Job导出数据

用法：

hbase org.apache.hadoop.hbase.mapreduce.Export 表名 路径

在webUI上查看结果