有时需要从hbase中一次读取大量的数据，同时对实时性有较高的要求。可以从两方面进行考虑：1、hbase提供的get方法提供了批量获取数据方法，通过组装一个list<Get> gets即可实现；2、java多线程的Future方法实现了如何从多线程中获取返回数据。以上两种方法结合后，获取数据将会更加的高效。阅读到一篇文章，对这两个方法的结合使用给出了实例，并有详细的性能分析。特意转载过来，供以后的参考学习：
如何保证在十亿、百亿数据上面的查询效率？ 答：使用分布式搜索引擎
数据量过亿，无论是存储在关系型数据库还是非关系型数据库，使用非索引字段进行条件查询、模糊查询等复杂查询都是一件极其缓慢甚至是不可能完成的任务，数据库索引建立的是二级索引，大数据查询主要依靠搜索引擎。
根据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基本说明与性能测试
从上面的测试结果表中可以看出，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的请求，并且采用多线程方式，经过验证，这种方法的效率还是蛮高的。
|
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;
ExecutorService pool = Executors.newFixedThreadPool(10);
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);
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();
}
finally
{
pool.shutdown();
}
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)
{
HTableInterface hTableInterface = null;
try
{
hTableInterface = getHConnection().getTable(tableName.getBytes());
}
catch (IOException e)
{
e.printStackTrace();
}
return hTableInterface;
} |
不得不提的是：在实现过程中，我曾将这种大批量请求化为每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要好得多。