快速上手NoSQL数据库HBase-3

5 HBase调优策略和扩展内容

HBase 调忧策略

预分区

HBase默认新建的表中只有一个Region，这个Region的Rowkey是没有边界的，即没有startRowkey和endRowkey，在数据写入时，所有数据都会写入这个默认的Region

随着数据量的不断增加，此Region已经不能承受不断增长的数据量，会进行Split，分裂成2个Region。
在这个过程中，会产生两个问题：
-数据往一个Region上写,会有写热点问题。
-Region split会消耗宝贵的集群IO资源。

基于此我们可以控制在建表的时候，创建多个空Region，并确定每个Region的起始和终止Rowkey，这样只要我们设计的Rowkey能均匀的命中各个Region，就不会存在写热点问题。Region分裂的几率也会大大降低。当然随着数据量的不断增长，该分裂还是要进行分裂的。

像这种预先给HBase表创建多个Region的方式，称之为预分区。
hbase(main):001:0> create 't20', 'c1', SPLITS => ['10', '20', '30', '40']
Created table t20
Took 3.3741 seconds                                                   
=> Hbase::Table - t20

1 2	到HBase的界面中查看这个表的region信息 http://bigdata01:16010/

通过这个图可以看出来，此时创建的这个表会提前创建多个Region。

默认情况下创建的表是只有1个Region的。
以表t1为例：

RowKey的设计原则

Rowkey长度原则

Rowkey底层存储是一个二进制流，可以是任意字符串，最大长度 64kb ，实际应用中一般是10-100字节，以byte[]形式保存，一般设计为定长。
建议越短越好，不要超过16个字节，原因如下：

-数据的持久化文件HFile中是按照KeyValue存储的，如果Rowkey过长，比如超过100字节，1000w行数据，Rowkey就要占用100*1000w=10亿字节，将近1G数据，这样会极大影响HFile的存储效率；

-MemStore会缓存部分数据到内存，如果Rowkey字段过长，内存的有效利用率就会降低，系统不能缓存更多的数据，这样会降低检索效率。

-目前操作系统都是64位系统，内存8字节对齐，控制在16个字节，8字节的整数倍利用了操作系统的最佳特性

Rowkey散列原则

Rowkey散列原则，主要是为了避免数据热点问题。
虽然我们可以在建表的时候提前设计预分区，但是假设数据的Rowkey都是手机号，那么都是1开头，按照前面的设计，那么所有的数据都会写到10-20之间的Region中，仍然没有做到负载均衡。
如何保证我们的数据能够均匀的分布到预先设计好的分区中呢？
解决思路（以手机号为例）：

-手机号反转，将手机号的最后一位前置，这样第一位就是0-9之间的任意一个数字了。
-按照一定规则使用hashCode获取余数，拼在手机号前面。例如：根据手机号后四位使用hashCode获取余数。这里的规则一定要是可以反推出来的，这样后期还可以根据这个规则找到对应的手机号，尽量不要使用随机数。

Rowkey唯一性原则

1
2
3

必须在设计上保证其唯一性，因为Rowkey相同则会覆盖
Rowkey是HBase里面唯一的索引,对于某些查询频繁的限定条件可以把它的内容存放在Rowkey里面，提高查询效率。
例如：需要经常使用姓名和年龄这两个字段进行查询，那么可以考虑把姓名和年龄拼接到一块作为Rowkey。

列族的设计原则

1
2

在设计列族的时候，建议把经常读取的字段存储到一个列族中，不经常读取的字段放到另一个列族中。
这样在读取部分数据的时候，就只需要读取一个列族文件即可，可以提高读取效率。

批量处理

Table.get(Get)方法可以根据一个指定的Rowkey获取一行记录，同样HBase提供了另一个方法：通过调用Table.get(List)方法可以根据一个指定的Rowkey列表，批量获取多行记录，这样做的好处是批量执行，只需要一次网络IO开销，这样可以带来明显的性能提升。
同理Table.delete(List) 和Table.put(List)

如果一次操作的数据量不是特别多，例如：100~1000条左右的数据量，可以考虑这种方式。
如果是一次需要批量操作上千万的数据，建议使用前面讲的批量导入导出方法，效率更高。

Region的request计数

HBase UI界面table Regions中的Requests参数值
这个参数的意义在于，可以分析哪个Region被频繁请求，是否存在读写热点的问题。

注意：HBase集群重启之后，Requests参数值会被清空。

以student表为例：

HBase核心参数优化

hbase.hregion.majorcompaction
配置大合并的间隔时间，默认为604800000毫秒(7天)，可设置为0，禁止自动的大合并，大合并的执行可能会持续数小时，为减少对业务的影响，建议在业务低峰期进行手动或者通过脚本或者API定期进行大合并。

hbase.hregion.max.filesize
默认为10737418240 Byte(10G)，当Region达到这个阈值时，会自动分裂。Region分裂会有短暂的Region下线时间(通常在5s以内)，为减少对业务端的影响，建议调大该值，并在业务低峰期定时手动进行分裂。

hbase.regionserver.handler.count
默认30，对于大负载的Put（达到了M范围）或是大范围的Scan操作，handler数目不易过大，易造成OOM（内存溢出）。 对于小负载的put、get，delete等操作，handler数要适当调大。handler属于一个处理器，实现底层数据的发送。

hbase.hregion.memstore.flush.size
默认值134217728 Byte (128M)，单位字节，这个参数是Memstore中数据持久化到Storefile的时机，超过该阈值，则会把Memstore中的数据持久化到Storefile中，如果Regionserver的JVM内存比较充足(例如：16G以上)，可以适当调大该值，例如：调整为256M。这样可以减少Memstore中数据溢写文件的次数。

hbase.hregion.memstore.block.multiplier
默认值4，如果一个Memstore的内存大小已经超过hbase.hregion.memstore.flush.size * hbase.hregion.memstore.block.multiplier，则会阻塞该Memstore的写操作，为避免阻塞，可以适当调大，例如6~8，但如果太大，则会有OOM的风险。 如果在Regionserver日志中出现"Blocking updates for ‘’ on region : memstore size <多少M> is >= than blocking <多少M> size"的信息时，说明这个值该调整了。

hbase.hstore.compaction.min
默认值为3，如果任何一个Store里的Storefile总数超过该值，会触发默认的合并操作，可以设置5~8，在手动的定期大合并中进行Storefile文件的合并，减少合并的次数，不过这会延长合并的时间

这些参数其实偏向于运维岗位的范畴，开发人员可以作为了解即可。
如果想要修改这些参数的话需要在hbase-site.xml中进行修改。

这些参数的默认值是在hbase-default.xml中的。

而hbase-default.xml文件在hbase-common-2.2.7.jar里面。

HBase扩展内容

【扩展】Hive与HBase 整合

Hive提供了与HBase的集成，可以在HBase表上使用HQL语句进行查询，插入操作以及进行join和union等复杂查询。

Hive整合HBase后的使用场景：

-通过Hive把数据加载到HBase中，数据源可以是文件也可以是Hive中的表。
-通过整合，让HBase支持JOIN、GROUP等SQL查询语法。
-通过整合，不仅可完成HBase的数据实时查询，也可以使用Hive查询HBase中的数据完成复杂的数据分析。

注意：Hive查询HBase中的数据，性能一般，并不能发挥HBase中根据Rowkey查询性能较高的特性。了解即可，实际工作中基本不会这样使用。

如果确实既有海量数据读写需求，还有SQL查询需求，可以考虑将数据存储两份，HBase中维护实时读写的数据，然后定时将数据导出到HDFS中，在Hive中映射表提供SQL查询服务。

【扩展】Phoenix（凤凰）

参考

Phoenix是构建在HBase上的一个SQL层，可以用标准的JDBC APIs来创建表，插入数据和对数据进行查询。

Phoenix完全使用Java编写，作为HBase内嵌的JDBC驱动。Phoenix查询引擎会将SQL查询转换为一个或多个HBase扫描，并编排执行以生成标准的JDBC结果集。直接使用HBase API、协同处理器与自定义过滤器，对于简单查询来说，其性能量级是毫秒，对于百万级别的行数来说，其性能量级是秒。

Phoenix通过以下方式使我们可以少写代码，并且性能比我们自己写代码更好：

-将SQL编译成原生的HBase scans。
-确定scan关键字的最佳开始和结束
-让scan并行执行

【扩展】协处理器coprocessor

参考

HBase作为列族数据库最经常被人诟病的特性包括：无法轻易建立“二级索引”，难以执行求和、计数、排序等操作。

比如，在旧版本的(<0.92)HBase中，统计数据表的总行数，需要使用Counter方法，执行一次MapReduce Job才能得到。虽然HBase在数据存储层中集成了MapReduce，能够有效用于数据表的分布式计算。然而在很多情况下，做一些简单的相加或者聚合计算的时候，如果直接将计算过程放置在server端，能够减少通讯开销，从而获得很好的性能提升。于是，HBase在0.92之后引入了协处理器(coprocessors)，实现一些激动人心的新特性：能够轻易建立二次索引、复杂过滤器以及访问控制等。

协处理器有两种：observer和 endpoint

-Observer 允许集群在正常的客户端操作过程中可以有不同的行为表现
-Endpoint 允许扩展集群的能力，对客户端应用开放新的运算命令
-Observer 类似于 RDBMS 中的触发器，主要在服务端工作
-Endpoint 类似于 RDBMS 中的存储过程，主要在服务端工作
-Observer 可以实现权限管理、优先级设置、监控、ddl 控制、二级索引等功能
-Endpoint 可以实现 min、max、avg、sum、distinct、group by 等功能

【扩展】Elasticsearch + HBase

HBase里面只有RowKey作为一级索引， 如果要对表里的非RowKey字段进行数据检索和查询， 往往要通过MapReduce/Spark等分布式计算框架进行，硬件资源消耗和时间延迟都会比较高。

由于HBase不支持多条件查询，不提供二级索引，难以满足用户对检索功能多样性和高效率两方面的需求。

本方案通过提出数据与索引的分离，利用HBase数据库的存储模式灵活多变，容纳海量数据等特点，结合Elasticsearch (简称为ES，ES是一个支持分布式的全文检索工具)的快速建立索引和提供多样化的查询接口等优势，构建基于ES的HBase二级索引方案。

思路：将索引数据存储于ES中，做查询时，先到ES中查询，转换为统一的RowKey后，再拿RowKey到HBase中快速定位。

注意：针对Elasticsearch +HBase架构设计会在后续更新的【仿百度搜索引擎】项目中实现。

【扩展】HBase实现分页功能

核心思路：
使用PageFilter过滤器+循环动态设置startRow实现

注意：循环动态设置startRow时，需要在上一次获取到的rowkey后面补0，表示新的开始，因为默认startRow中包含这一行数据

正常情况PageFilter返回的结果数量可能大于设定的值，因为服务器集群的PageFilter是隔离的，只能保证每个Region返回的数据量不会超过PageFilter中设置的值。

注意：想要解决返回数据超过设置数量的问题，可以考虑对获取到的数据进行截取，只保留需要的数据条数，下一次读取时根据上次截取的位置开始往后读取。
当PageFilter和其它Filter一起使用时，需要将PageFilter加入到FilterList的末尾，否则会出现结果个数小于你期望的数量。

首先创建一个带有预分区的表，并且向表里面初始化一批测试数据。
create 'test', 'info','date', SPLITS => ['10', '20', '30', '40']

put 'test','10001','info:name','zs'
put 'test','10001','info:address','bj'
put 'test','10001','date:start_time','2021-01-01'
put 'test','10001','date:end_time','2021-01-01'

put 'test','10002','info:name','ww'
put 'test','10002','info:address','sh'
put 'test','10002','date:start_time','2021-01-02'
put 'test','10002','date:end_time','2021-01-02'

put 'test','20001','info:name','ls'
put 'test','20001','info:address','hz'
put 'test','20001','date:start_time','2021-01-01'
put 'test','20001','date:end_time','2021-01-01'

put 'test','20002','info:name','jack'
put 'test','20002','info:address','sh'
put 'test','20002','date:start_time','2021-01-02'
put 'test','20002','date:end_time','2021-01-02'

put 'test','30001','info:name','ls'
put 'test','30001','info:address','hz'
put 'test','30001','date:start_time','2021-01-01'
put 'test','30001','date:end_time','2021-01-01'

put 'test','30002','info:name','tom'
put 'test','30002','info:address','lz'
put 'test','30002','date:start_time','2021-01-02'
put 'test','30002','date:end_time','2021-01-02'

代码如下：

package com.imooc.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

/**
 * 结合Scan实现分页功能
 * Created by xuwei
 */
public class PageFilterOp {
    public static void main(String[] args) throws Exception{
        //获取配置
        Configuration conf = HBaseConfiguration.create();
        //指定HBase使用的zk的地址，多个都逗号隔开
        conf.set("hbase.zookeeper.quorum", "bigdata01:2181,bigdata02:2181,bigdata03:2181");
        //指定HBase在hdfs上的根目录
        conf.set("hbase.rootdir","hdfs://bigdata01:9000/hbase");
        //创建HBase连接，负责对HBase中数据的增删改查(DML操作)
        Connection conn = ConnectionFactory.createConnection(conf);

        //获取到test表的链接
        Table table = conn.getTable(TableName.valueOf("test"));

        Scan scan = new Scan();

        //组装Filter列表
        FilterList filterList = new FilterList();

        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("name"), CompareOperator.EQUAL, new SubstringComparator("w"));
        //注意：如果用到了多个filter，其中包含pagefilter，那么pagefilter需要放在fiterlist的最后一个
        filterList.addFilter(singleColumnValueFilter);

        //每页的数据量
        int pageSize = 2;
        //注意：pagefilter返回的数据总数会超过这里设置的数量，需要自己在程序内部处理
        PageFilter pageFilter = new PageFilter(pageSize);
        filterList.addFilter(pageFilter);

        //给scan设置Filter
        scan.setFilter(filterList);

        //记录上一次返回的分页数据中的最大的Rowkey，最开始为null
        byte[] lastRowKey = null;//Bytes.toBytes("10002");

        if(lastRowKey == null){//第一次查询，查询第一页数据
            //设置一个不存在的Rowkey
            scan.withStartRow(Bytes.toBytes("-1"));
        }else{//查询第二页或者其他页数的数据
            //注意：在这里需要在lastRowkey后面补0，否则会把当前这条数据也返回过来，这样就重复了，补0之后可以保证返回的都是新数据
            scan.withStartRow(Bytes.add(lastRowKey,Bytes.toBytes("0")));
        }
        //记录每次迭代的数据条数
        int rowCount = 0;
        //获取结果
        ResultScanner scanner = table.getScanner(scan);
        for (Result res: scanner) {
            byte[] rowkey_bytes = res.getRow();
            byte[] name_bytes = res.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"));
            String rowkey_str = new String(rowkey_bytes);
            String name_str = new String(name_bytes);
            System.out.println("Rowkey:"+ rowkey_str +",name:"+ name_str);
            rowCount++;
            //scan返回的数据是基于rowkey有序的，直接判断数据条数即可。
            if (rowCount == pageSize) {//当前页面数据获取完毕，退出循环
                System.out.println("lastRowKey:"+rowkey_str);
                break;
            }
        }
    }
}

【扩展】封装HBaseUtils工具类

要求：

通过静态方法维护HBase连接。
提供公共的put和get方法。
代码如下

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

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * HBase工具类
 */
public class HBaseUtil {
    private HBaseUtil(){}

    private static Connection conn  = getConn();

    private static Connection getConn(){
        //获取hbase的链接
        Configuration conf = new Configuration();
        //指定hbase使用的zk地址
        //注意：需要在执行hbasejava代码的机器上配置zk和hbase集群的主机名和ip的映射关系
        conf.set("hbase.zookeeper.quorum","bigdata01:2181");
        //指定hbase在hdfs上的根目录
        conf.set("hbase.rootdir","hdfs://bigdata01:9000/hbase");
        //创建HBase数据库链接
        Connection co = null;
        try{
            co = ConnectionFactory.createConnection(conf);
        }catch (IOException e){
            System.out.println("获取链接失败："+e.getMessage());
        }
        return co;
    }

    /**
     * 对外提供的方法
     * @return
     */
    public static Connection getInstance(){
        return conn;
    }

    /**
     * 创建表
     * @param tableName
     * @param cfs
     * @throws Exception
     */
    public static void createTable(String tableName,String... cfs)throws Exception{
        Admin admin = conn.getAdmin();
        ArrayList<ColumnFamilyDescriptor> cfArr = new ArrayList<>();
        for (String cf: cfs) {
            ColumnFamilyDescriptor cfDesc = ColumnFamilyDescriptorBuilder
                    .newBuilder(Bytes.toBytes(cf))
                    .build();
            cfArr.add(cfDesc);
        }
        TableDescriptor tableDesc = TableDescriptorBuilder
                .newBuilder(TableName.valueOf(tableName))
                .setColumnFamilies(cfArr)
                .build();
        admin.createTable(tableDesc);
        admin.close();
    }

    /**
     * 添加一个单元格(列)的数据
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @param value
     * @throws Exception
     */
    public static void put2HBaseCell(String tableName,String rowKey,String columnFamily,String column,String value) throws Exception{
        Table table = conn.getTable(TableName.valueOf(tableName));
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(column),Bytes.toBytes(value));
        table.put(put);
        table.close();
    }

    /**
     * 向hbase中添加一批数据
     * @param tableName
     * @param list
     * @throws Exception
     */
    public static void put2HBaseList(String tableName, List<Put> list)throws Exception{
        Table table = conn.getTable(TableName.valueOf(tableName));
        table.put(list);
        table.close();
    }

    public static void main(String[] args) throws Exception{
        HBaseUtil.put2HBaseCell("user","1002008","info","name","abc");
    }

}

HBase常见问题总结

-HBase Put 功能初始化数据过慢，考虑使用批量导入。
-统一各个系统的字符集，非utf8的要做转换。
-对表做预分区，同时Rowkey做MD5哈希取余数。
-在HBase客户端节点中需要配置HBase集群所有节点的主机名和IP的映射关系。
-每日全量数据入库，数据实际发生变化的条数不多，浪费资源，所以用T-2的数据和T-1的数据做对比，只入库发生变化的数据。
-Scan大表超时，最好限制一个范围，尝试调整RPC请求的超时时间，hbase.rpc.timeout，可以适当调大。
-默认建表version=1，手动修改version=3，可以查找之前修改的记录。
-HBase第一次查询数据很慢，建议提前初始化链接。

比你优秀的人都努力，有什么理由不努力！

大数据开发工程师-快速上手NoSQL数据库HBase-3

快速上手NoSQL数据库HBase-3

5 HBase调优策略和扩展内容

HBase 调忧策略

预分区

RowKey的设计原则

Rowkey长度原则

Rowkey散列原则

Rowkey唯一性原则

列族的设计原则

批量处理

Region的request计数

HBase核心参数优化

HBase扩展内容

【扩展】Hive与HBase 整合

【扩展】Phoenix（凤凰）

【扩展】协处理器coprocessor

【扩展】Elasticsearch + HBase

【扩展】HBase实现分页功能

【扩展】封装HBaseUtils工具类

HBase常见问题总结