HBase 高性能获取数据(多线程批量式解决办法) + MySQL和HBase性能测试比较

2021-12-23 22:47:01

摘要：在前篇博客里已经讲述了通过一个自定义 HBase Filter来获取数据的办法，在末尾指出此办法的性能是不能满足应用要求的，很显然对于如此成熟的HBase来说，高性能获取数据应该不是问题。下面首先简单介绍了搜索引擎的性能，然后详细说明了HBase与MySQL的性能对比，这里的数据都是经过实际的测试获得的。最后，给出了采用多线程批量从HBase中取数据的方案，此方案经过测试要比通过自定义Filter的方式性能高出很多。

关键词：　HBase, 高性能, 获取数据, 性能对比, 多线程

需求：　　 HBase高性能获取数据

Solr和HBase专辑

1、“关于Solr的使用总结的心得体会”(http://www.cnblogs.com/wgp13x/p/3742653.html)

2、“中文分词器性能比较”(http://www.cnblogs.com/wgp13x/p/3748764.html)

3、“Solr与HBase架构设计”(http://www.cnblogs.com/wgp13x/p/a8bb8ccd469c96917652201007ad3c50.html)

4、 “大数据架构: 使用HBase和Solr将存储与索引放在不同的机器上”(http://www.cnblogs.com/wgp13x/p/3927979.html)

5、“一个自定义 HBase Filter -通过RowKeys来高性能获取数据”(http://www.cnblogs.com/wgp13x/p/4196466.html)

6、“HBase 高性能获取数据 - 多线程批量式解决办法”（http://www.cnblogs.com/wgp13x/p/4245182.html）

1、如何存储十亿、百亿数据？答：使用数据存储集群，增加水平拓展能力，以容纳上百亿数据量

2、如何保证在十亿、百亿数据上面的查询效率？答：使用分布式搜索引擎

数据量过亿，无论是存储在关系型数据库还是非关系型数据库，使用非索引字段进行条件查询、模糊查询等复杂查询都是一件极其缓慢甚至是不可能完成的任务，数据库索引建立的是二级索引，大数据查询主要依靠搜索引擎。

根据Solr中国资料显示，在2400亿每条数据大概200字节的数据建立索引，搭建分布式搜索引擎，在50台机器进行搜索测试，其中有条件查询、模糊查询等，其中80%的搜索能够在毫秒内返回结果，剩下一部分能够在20秒内返回，还有5%左右的查询需要在50秒左右的时间完成查询请求，客户端查询请求的并发量为100个客户端。

以下结论均是在同一台服务器上的测试结果。

MySQL单机随机读写能力测试

MySQL(InnoDB)
运行环境	Window Server 2008 x64
存储引擎	InnoDB
最大存储容量	64T
列数	39列
每条数据的大小	Avg=507Byte
总数据量	302,418,176条
占用的磁盘空间	210G
插入效率	总共耗时13个小时，每秒约6500条，随着数据量的增大，插入的效率影响不大
单条数据全表随机读取时间	30ms
百条数据全表随机读取时间	1,783ms;1,672ms
千条数据全表随机读取时间	18,579ms;15,473ms
其他	条件查询、Order By、模糊查询基本上是无法响应的

HBase基本说明与性能测试

HBase
数据库类型	NoSql—列式数据库
运行所需要的环境	Linux
是否可以搭建集群	天然的分布式数据库，具有自动分片功能
可扩展性	强，无缝支持水平拓展
插入	与设置的参数关系很大，批量插入和单条插入差别大，单台机器能够实现1w~3w之间的插入速度
更新
删除
查询	只支持按照rowkey来查询或者全表扫描
范围查询	不支持
模糊匹配	不支持
时间范围查询	不支持
分页查询	可以做到
数据库安全性	低
大数据量下的查询响应时间	各个数据级别下的响应时间： (均为随机读取，不命中缓存) 1、3亿-------------------5ms(单行) 2、3亿-------------------124ms(30行)
大数据量下占用的磁盘空间	各个数据级别下的磁盘占用空间（以出租车表为例，17个字段，一行200个字节）： 1、1亿-------------------18G（使用GZ压缩）
是否有良好的技术支持	社区活跃，但是配置复杂，参数繁多，学习代价比较大
数据导入和导出	有从RDBMS导入数据的工具Sqoop
热备份
异步复制
是否需要商业付费	否
是否开源	是
优点	1、支持高效稳定的大数据存储，上亿行、上百万列、上万个版本，对数据自动分片 2、列式存储保证了高效的随机读写能力 3、列数可以动态增长 4、水平拓展十分容易 5、拥有良好的生态系统，Sqoop用户数据的导入、Pig可以作为ETL工具，Hadoop作为分布式计算平台
缺点	1、学习复杂 2、不支持范围查询、条件查询等查询操作

从上面的测试结果表中可以看出，MySQL单表插入速度为每秒6500条，HBase单台机器能够实现1w~3w之间的插入速度，这充分说明HBase插入数据的速度比MySQL高很多。在MySQL单机随机读写能力测试中，单条数据全表随机读取时间是指依据主键去MySQL单表取数据花费的时间；在HBase基本说明与性能测试中，大数据量下查询响应时间是指依照Rowkey到HBase取数据所花费的时间。30ms对5ms，这说明HBase取数据的速度之快也是MySQL望尘莫及的。

在进行上面的性能测试中，无论是从MySQL通过主键读取，还是从HBase通过Rowkey读取，读取的数据量都不大，不超过1000条。当需要一次性读取万级数据时，需要通过设计优化的代码来保证读取速度。

在实现过程中，发现当批量Get的数据量达到一定程度时（如10W），向HBase请求数据会从innerGet发生EOFExeption异常。这里附加上一段从HBase依照多Rowkey获取数据的代码，它采用了性能高的批量Get。在这里，我将这种大批量请求化为每1000个Get的请求，并且采用多线程方式，经过验证，这种方法的效率还是蛮高的。

private ExecutorService pool = Executors.newFixedThreadPool(10);    // 这里创建了10个 Active RPC Calls

    public Datas getDatasFromHbase(final List<String> rowKeys,

        final List<String> filterColumn, boolean isContiansRowkeys,

        boolean isContainsList)

    {

        if (rowKeys == null || rowKeys.size() <= 0)

        {

            return Datas.getEmptyDatas();

        }

        final int maxRowKeySize = 1000;

        int loopSize = rowKeys.size() % maxRowKeySize == 0 ? rowKeys.size()

            / maxRowKeySize : rowKeys.size() / maxRowKeySize + 1;

        ArrayList<Future<List<Data>>> results = new ArrayList<Future<List<Data>>>();

        for (int loop = 0; loop < loopSize; loop++)

        {

            int end = (loop + 1) * maxRowKeySize > rowKeys.size() ? rowKeys

                .size() : (loop + 1) * maxRowKeySize;

            List<String> partRowKeys = rowKeys.subList(loop * maxRowKeySize,

                end);

            HbaseDataGetter hbaseDataGetter = new HbaseDataGetter(partRowKeys,

                filterColumn, isContiansRowkeys, isContainsList);

            synchronized (pool)

            {

                Future<List<Data>> result = pool.submit(hbaseDataGetter);

                results.add(result);

            }

        }

        Datas datas = new Datas();

        List<Data> dataQueue = new ArrayList<Data>();

        try

        {

            for (Future<List<Data>> result : results)

            {

                List<Data> rd = result.get();

                dataQueue.addAll(rd);

            }

            datas.setDatas(dataQueue);

        }

        catch (InterruptedException | ExecutionException e)

        {

            e.printStackTrace();

        }

        return datas;

    }

class HbaseDataGetter implements Callable<List<Data>>

{

private List<String> rowKeys;

private List<String> filterColumn;

private boolean isContiansRowkeys;

private boolean isContainsList;

public HbaseDataGetter(List<String> rowKeys, List<String> filterColumn,

boolean isContiansRowkeys, boolean isContainsList)

{

this.rowKeys = rowKeys;

this.filterColumn = filterColumn;

this.isContiansRowkeys = isContiansRowkeys;

this.isContainsList = isContainsList;

}

@Override

public List<Data> call() throws Exception

{

Object[] objects = getDatasFromHbase(rowKeys, filterColumn);

List<Data> listData = new ArrayList<Data>();

for (Object object : objects)

{

Result r = (Result) object;

Data data = assembleData(r, filterColumn, isContiansRowkeys,

isContainsList);

listData.add(data);

}

return listData;

}

private Object[] getDatasFromHbase(List<String> rowKeys,

List<String> filterColumn)

{

createTable(tableName);

Object[] objects = null;

HTableInterface hTableInterface = createTable(tableName);

List<Get> listGets = new ArrayList<Get>();

for (String rk : rowKeys)

{

Get get = new Get(Bytes.toBytes(rk));

if (filterColumn != null)

{

for (String column : filterColumn)

{

get.addColumn(columnFamilyName.getBytes(),

column.getBytes());

}

listGets.add(get);

}

try

{

objects = hTableInterface.get(listGets);

}

catch (IOException e1)

{

e1.printStackTrace();

}

finally

{

try

{

listGets.clear();

hTableInterface.close();

}

catch (IOException e)

{

e.printStackTrace();

}

return objects;

}

private HTableInterface createTable(String tableName)

    {

        HTable table = null;

        try

        {

            table = new HTable(initHbaseConfiguration(), tableName);

        }

        catch (IOException e)

        {

            e.printStackTrace();

        }

        return table;

    }

可以肯定的是，此种批量取数据的方法达成的速度，与取一次性数据的数量基本成线性关系，与总数据量相关不大，需要取出的数据越多耗时也就越多，经过测试一次性取1000条数据花费大约在2至3s以内，总数据量为400W。而通过自定义Filter方式取数据的方法的速度，与取一次性数据的数量相关不大，与总数据量成线性关系，总数据量越大取出越慢，即使只需取一条，因为此方式对HBase每条数据都过滤一遍。这样，如果在总数不大，需要取数据量较大的情况下，通过自定义Filter取数据的方式可能还占有些优势，但在正常情况下，此种批量取数据的方法还是优势更大。

不得不提的是：在实现过程中，我曾将这种大批量请求化为每4000个Get的多线程请求方式，我们的HBase版本为0.94，这样在一次性请求200000条数据时，HBase直接挂机，client抛出EOFException异常，【processBatchCallback(HConnectionManager.java:1708)，processBatch(HConnectionManager.java:1560)，(HTable.java:779)】，查看并发连接数与每1000个Get请求一样保持为10个左右，没有异常。查阅相关资料后，我们怀疑，这是由于HTable的非线程安全特性导致的，但经过多时纠缠，最终也没得到可靠结论。后来确定这是由于HBase0.94版自身的问题，在使用0.96版后，此问题便不再出现了。而且我们发现0.94版HBase并不稳定，经常有挂掉情况出现。0.96版HBase要好得多。

这里补充非常重要的一点，在上面的代码中，我通过 private ExecutorService pool = Executors.newFixedThreadPool(10); 创建了一个最多容纳10个线程的线程池，从而创建了10个 Active RPC Calls，有效提高了获取速度。然而，我将此线程池容量扩大至20个后，的确创建了20个 Active RPC Calls，如下图所示，但是会直接引起事故：HBase挂掉。不得不吐cao，HBase实在不稳定，维护极其花费成本。在种种实践验证后，才得到了这个稳定高效的方式，每1000个Get一次批量请求，至多10个线程同时取。

平均效率如下图所示：

更多的HBase及其它的数据存储方案测试情况，HBase高性能插入数据解决方案，正在整理中，敬请批评指正。

来自王安琪

码农公寓

相关文章