hbase-0.92.1-cdh4.1.3的HTablePool实现:
1. PoolType:
Reusable: (默认)一个实例池,多线程复用,内部用ConcurrentLinkedQueue装多个实例HTable;
ThreadLocal: 每个线程只有一个实例,线程与线程之间互不影响, ThreadLocal;
特点是随着线程的增多,Pool中的HTable增多,但互不影响;
RoundRobin: Pool中的HTable用CopyOnWriteArrayList装;
2. 初始化:
HTablePool pool = new HTablePool(conf, 5);
// 默认 PoolType.Reusable
pool = new HTablePool(conf, maxSize, tableFactory, PoolMap.PoolType.ThreadLocal);
// PoolMap.PoolType.ThreadLocal
pool = new HTablePool(conf, maxSize, tableFactory, PoolMap.PoolType.RoundRobin);
// PoolMap.PoolType.RoundRobin
实例化PoolMap
实例化HTablePool,此时还没有任何HTable实例,tables为空;
3. 取得HTableInterface对象:
pool.getTable(TEST_TABLE_NAME);
查看tables是否含有table,如果没有,创建一个HTable实例
返回HTable实例封装成PooledHTable实例返回
PooledHTable实例用完.close();后放置到PoolMap;
4. HTablePool可以容纳任何的Table的HTableInterface实例.
HTable实例会共享同一个zookeeper连接
HTable实例,如果同在一个RegionServer会共享同一个连接HBaseClient$Connection
HTablePool有最大尺寸,但并没有限制HTable实例不得大于这个尺寸,一旦超过这个尺寸就会实例化,但归还到实例池的时候,如果池满了会弃用;
HTable实例线程不安全;
注意点:
1. 在多线程使用HTablePool拿到同一个表的HTable时,如果线程个数大于maxsize会导致写入始终是autoflush!
public HTableInterface getTable(String tableName) { // call the old getTable implementation renamed to findOrCreateTable HTableInterface table = findOrCreateTable(tableName); // return a proxy table so when user closes the proxy, the actual table // will be returned to the pool return new PooledHTable(table); } public void close() throws IOException { returnTable(table); } private void returnTable(HTableInterface table) throws IOException { // this is the old putTable method renamed and made private String tableName = Bytes.toString(table.getTableName()); if (tables.size(tableName) >= maxSize) { // release table instance since we're not reusing it this.tables.remove(tableName, table); this.tableFactory.releaseHTableInterface(table); return; } tables.put(tableName, table); }
如果tables.size大于maxsize,此时会去掉一个保存的HTable对象,而releaseHTableInterface实际调用的就是HTable的close方法,close方法又会强制flushHTable的buffer,因此,如果我们想不使用autoflush提升写入速度失效。
2. 改写HTable,的flushCommit(固定频率+内存占用>1M)
@Override public void put(final List<Put> puts) throws IOException { super.put(puts); needFlush(); } private void needFlush() throws IOException { long currentTime = System.currentTimeMillis(); if ((currentTime - lastFlushTime.longValue()) > flushInterval) { super.flushCommits(); lastFlushTime.set(currentTime); } }
初始化使用代码样例
import java.io.IOException; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.FutureTask; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Get; import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.HTablePool; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.util.Bytes; import org.junit.BeforeClass; import org.junit.Test; public class HTablePoolTest2 { protected static String TEST_TABLE_NAME = "testtable"; protected static String ROW1_STR = "row1"; protected static String COLFAM1_STR = "colfam1"; protected static String QUAL1_STR = "qual1"; private final static byte[] ROW1 = Bytes.toBytes(ROW1_STR); private final static byte[] COLFAM1 = Bytes.toBytes(COLFAM1_STR); private final static byte[] QUAL1 = Bytes.toBytes(QUAL1_STR); private static HTablePool pool; @BeforeClass public static void runBeforeClass() throws IOException { Configuration conf = HBaseConfiguration.create(); // 默认使用PoolType.Reusable pool = new HTablePool(conf, 10); // 初始化填充pool HTableInterface[] tables = new HTableInterface[10]; for (int n = 0; n < 10; n++) { tables[n] = pool.getTable(TEST_TABLE_NAME); } // close后,PooledTable就放回了pool for (HTableInterface table : tables) { table.close(); } } @Test public void testHTablePool() throws IOException, InterruptedException, ExecutionException { Callable<Result> callable = new Callable<Result>() { public Result call() throws Exception { return get(); } }; FutureTask<Result> task1 = new FutureTask<Result>(callable); FutureTask<Result> task2 = new FutureTask<Result>(callable); Thread thread1 = new Thread(task1, "THREAD-1"); thread1.start(); Thread thread2 = new Thread(task2, "THREAD-2"); thread2.start(); Result result1 = task1.get(); System.out.println(Bytes.toString(result1.getValue(COLFAM1, QUAL1))); Result result2 = task2.get(); System.out.println(Bytes.toString(result2.getValue(COLFAM1, QUAL1))); } private Result get() { HTableInterface table = pool.getTable(TEST_TABLE_NAME); Get get = new Get(ROW1); try { Result result = table.get(get); return result; } catch (IOException e) { e.printStackTrace(); return null; } finally { try { table.close(); } catch (IOException e) { e.printStackTrace(); } } } }
http://blog.csdn.net/mrtitan/article/details/8892815
http://helpbs.iteye.com/blog/1492054
相关推荐
在HBase 0.92.1的安装过程中,用户首先需要解压`hbase-0.92.1.tar.gz`,然后配置相关环境变量。接着,安装并配置Zookeeper 3.3.5,解压`zookeeper-3.3.5.tar.gz`,配置`conf/zoo.cfg`文件,并启动Zookeeper服务。...
hbase-client-2.1.0-cdh6.3.0.jar
`hbase-1.2.0-cdh5.14.2.tar.gz` 是针对Cloudera Distribution Including Apache Hadoop (CDH) 5.14.2的一个特定版本的HBase打包文件。CDH是一个流行的Hadoop发行版,包含了多个大数据组件,如HDFS、MapReduce、YARN...
"hbase-1.2.0-cdh5.14.0.tar.gz" 是针对Cloudera Distribution Including Apache Hadoop (CDH) 5.14.0版本的HBase安装包,该版本的HBase是1.2.0版,与CDH 5.x系列兼容。 1. **HBase概述** HBase是一种非关系型...
hbase-1.2.0-cdh5.9.3.tar.gz
HBase 0.92.1是该系统的早期版本,尽管现在已经有更新的版本,但这个版本在当时仍然包含了众多重要的功能和改进。 HBase基于谷歌的Bigtable论文设计,它将数据存储在多维稀疏矩阵中,这种矩阵由行、列和时间戳组成...
此压缩包"**hbase-0.98.6-cdh5.3.6.zip**"包含了针对CDH(Cloudera Distribution Including Apache Hadoop)5.3.6版本优化的HBase 0.98.6版本。CDH是由Cloudera公司提供的一个开源大数据平台,集成了包括Hadoop在内...
hbase-0.98.6-cdh5.3.6.rarhbase-0.98.6-cdh5.3.6.rarhbase-0.98.6-cdh5.3.6.rarhbase-0.98.6-cdh5.3.6.rarhbase-0.98.6-cdh5.3.6.rarhbase-0.98.6-cdh5.3.6.rarhbase-0.98.6-cdh5.3.6.rarhbase-0.98.6-cdh5.3.6....
hbase-0.98.6-cdh5.3.6-src.rarhbase-0.98.6-cdh5.3.6-src.rarhbase-0.98.6-cdh5.3.6-src.rarhbase-0.98.6-cdh5.3.6-src.rarhbase-0.98.6-cdh5.3.6-src.rarhbase-0.98.6-cdh5.3.6-src.rarhbase-0.98.6-cdh5.3.6-src....
HBase 元数据修复工具包。 ①修改 jar 包中的application.properties,重点是 zookeeper.address、zookeeper.nodeParent、hdfs....③开始修复 `java -jar -Drepair.tableName=表名 hbase-meta-repair-hbase-2.0.2.jar`
hbase-sdk是基于hbase-client和hbase-thrift的原生API封装的一款轻量级的HBase ORM框架。 针对HBase各版本API(1.x~2.x)间的差异,在其上剥离出了一层统一的抽象。并提供了以类SQL的方式来读写HBase表中的数据。对...
赠送jar包:phoenix-core-4.7.0-HBase-1.1.jar; 赠送原API文档:phoenix-core-4.7.0-HBase-1.1-javadoc.jar; 赠送源代码:phoenix-core-4.7.0-HBase-1.1-sources.jar; 赠送Maven依赖信息文件:phoenix-core-4.7.0...
HBaseContext所需要的jar包。由于不知道为啥maven配置仓库地址不生效。所以手动下载了个。导入本地环境的maven仓库。
标题“hbase-1.2.1-bin.tar.gz.zip”表明这是HBase 1.2.1版本的二进制发行版,以tar.gz格式压缩,并且进一步用zip压缩。这种双重压缩方式可能用于减小文件大小,方便在网络上传输。用户需要先对zip文件进行解压,...
HBase(hbase-2.4.9-bin.tar.gz)是一个分布式的、面向列的开源数据库,该技术来源于 Fay Chang 所撰写的Google论文“Bigtable:一个结构化数据的分布式存储系统”。就像Bigtable利用了Google文件系统(File System...
HBase是参考google的bigtable的一个开源产品,建立在hdfs之上的一个提供高可靠性、高性能、列存储、可伸缩、实时读写的数据库系统。是一种介于nosql和RDBMs之间的一种数据库系统,仅支持通过rowkey和range进行数据的...
phoenix-client-hbase-2.2-5.1.2.jar
"phoenix-5.0.0-HBase-2.0-client" 是一个针对Apache HBase数据库的Phoenix客户端库,主要用于通过SQL查询语句与HBase进行交互。这个版本的Phoenix客户端是为HBase 2.0版本设计和优化的,确保了与该版本HBase的兼容...
本文将围绕"Hbase-1.2.0-cdh5.12.0.tar.gz"这一版本,深入探讨HBase在CDH(Cloudera Distribution Including Apache Hadoop)5.12.0环境下的具体应用、特点以及与其他组件如Hadoop、Hive、Kafka的协同工作。...