图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Hbase › 连载型› hbase源码系列（二）HTable 探秘
分享

hbase源码系列（二）HTable 探秘

坎蒂丝_Swan 2014-12-24 16:36:15 发表于 连载型 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 0 24255
About云VIP帮助找工作和提升
本帖最后由 坎蒂丝_Swan 于 2014-12-29 12:55 编辑
问题导读
１．HTable内部是用什么来和后台通信的？
２．有哪些客户端调优的方法？






hbase的源码终于搞一个段落了，在接下来的一个月，着重于把看过的源码提炼一下，对一些有意思的主题进行分享一下。继上一篇讲了负载均衡之后，这一篇我们从client开始讲吧，从client到master再到region server，按照这个顺序来开展，网友也可以对自己感兴趣的部分给我留言或者直接联系我的QQ。
　　现在我们讲一下HTable吧，为什么讲HTable，因为这是我们最常见的一个类，这是我们对hbase中数据的操作的入口。

1.Put操作

　　下面是一个很简单往hbase插入一条记录的例子。

  1. HBaseConfiguration conf =  (HBaseConfiguration) HBaseConfiguration.create();

  2. byte[] rowkey = Bytes.toBytes("cenyuhai");

  3. byte[] family = Bytes.toBytes("f");

  4. byte[] qualifier = Bytes.toBytes("name");

  5. byte[] value = Bytes.toBytes("岑玉海");

  6.         

  7. HTable table = new HTable(conf, "test");

  8. Put put = new Put(rowkey);

  9. put.add(family,qualifier,value);

  10.         

  11. table.put(put);
复制代码


我们平常就是采用这种方式提交的数据，为了提高重用性采用HTablePool，最新的API推荐使用HConnection.getTable("test")来获得HTable，旧的HTablePool已经被抛弃了。好，我们下面开始看看HTable内部是如何实现的吧，首先我们看看它内部有什么属性。
  1. /** 实际提交数据所用的类 */

  2.   protected HConnection connection;/** 需要提交的数据的列表 */

  3.   protected List<Row> writeAsyncBuffer = new LinkedList<Row>();

  4.   /** flush的size */

  5.   private long writeBufferSize;

  6.   /** 是否自动flush */

  7.   private boolean autoFlush;

  8.   /** 当前的数据的size，达到指定的size就要提交 */

  9.   protected long currentWriteBufferSize;

  10.   protected int scannerCaching;

  11.   private int maxKeyValueSize;

  12.   private ExecutorService pool;  // For Multi

  13. 

  14.   /** 异步提交 */

  15.   protected AsyncProcess<Object> ap;

  16.   ** rpc工厂 */

  17.   private RpcRetryingCallerFactory rpcCallerFactory;
复制代码


主要是靠上面的这些家伙来干活的，这里面的connection、ap、rpcCallerFactory是用来和后台通信的，HTable只是做一个操作，数据进来之后，添加到writeAsyncBuffer，满足条件就flush。

下面看看table.put是怎么执行的：

  1.  doPut(put);

  2.     if (autoFlush) {

  3.       flushCommits();

  4.     }
复制代码

执行put操作，如果是autoFush，就提交，先看doPut的过程，如果之前的ap异步提交到有问题，就先进行后台提交，不过这次是同步的，如果没有错误，就把put添加到队列当中，然后检查一下当前的 buffer的大小，超过我们设置的内容的时候，就flush掉。
  1. if (ap.hasError()){

  2.       backgroundFlushCommits(true);

  3. }

  4. currentWriteBufferSize += put.heapSize();

  5. writeAsyncBuffer.add(put);

  6. while (currentWriteBufferSize > writeBufferSize) {

  7.     backgroundFlushCommits(false);

  8. }
复制代码


写下来，让我们看看backgroundFlushCommits这个方法吧，它的核心就这么一句ap.submit(writeAsyncBuffer, true) ，如果出错了的话，就报错了。所以网上所有关于客户端调优的方法里面无非就这么几种:
1)关闭autoFlush
2)关闭wal日志
3)把writeBufferSize设大一点，一般说是设置成5MB

　　经过实践，就第二条关闭日志的效果比较明显，其它的效果都不明显，因为提交的过程是异步的，所以提交的时候占用的时间并不多，提交到server端后，server还有一个写入的队列，(⊙o⊙)… 让人想起小米手机那恶心的排队了。。。所以大规模写入数据，别指望着用put来解决。。。mapreduce生成hfile，然后用bulk load的方式比较好。

　　不废话了，我们继续追踪ap.submit方法吧，F3进去。

  1. int posInList = -1;

  2.       Iterator<? extends Row> it = rows.iterator();

  3.       while (it.hasNext()) {

  4.         Row r = it.next();

  5.         //为row定位

  6.         HRegionLocation loc = findDestLocation(r, 1, posInList);

  7. 

  8.         if (loc != null && canTakeOperation(loc, regionIncluded, serverIncluded)) {

  9.           // loc is null if there is an error such as meta not available.

  10.           Action<Row> action = new Action<Row>(r, ++posInList);

  11.           retainedActions.add(action);

  12.           addAction(loc, action, actionsByServer);

  13.           it.remove();

  14.         }

  15.       }
复制代码
　循环遍历r，为每个r找到它的位置loc，loc是HRegionLocation，里面记录着这行记录所在的目标region所在的位置，loc怎么获得呢，走进findDestLocation方法里面，看到了这么一句。
  1. loc = hConnection.locateRegion(this.tableName, row.getRow());
复制代码


通过表名和rowkey，使用HConnection就可以定位到它的位置，这里就先不讲定位了，稍后放一节出来讲，请看这一篇《Client如何找到正确的Region Server》，否则篇幅太长了，这里我们只需要记住，提交操作，是要知道它对应的region在哪里的。

　　定位到它的位置之后，它把loc添加到了actionsByServer，一个region server对应一组操作。（插句题外话为什么这里叫action呢，其实我们熟知的Put、Delete，以及不常用的Append、Increment都是继承自Row的，在接口传递时候，其实都是视为一种操作，到了后台之后，才做区分）。

　　接下来，就是多线程的rpc提交了。

  1. MultiServerCallable<Row> callable = createCallable(loc, multiAction);

  2. ......

  3. res = createCaller(callable).callWithoutRetries(callable);
复制代码

再深挖一点，把它们的实现都扒出来吧。
  1. protected MultiServerCallable<Row> createCallable(final HRegionLocation location,

  2.       final MultiAction<Row> multi) {

  3.     return new MultiServerCallable<Row>(hConnection, tableName, location, multi);

  4.   }

  5. 

  6.   protected RpcRetryingCaller<MultiResponse> createCaller(MultiServerCallable<Row> callable) {

  7.     return rpcCallerFactory.<MultiResponse> newCaller();

  8.   }
复制代码


　　ok，看到了，先构造一个MultiServerCallable，然后再通过rpcCallerFactory做最后的call操作。
　　好了，到这里再总结一下put操作吧，前面写得有点儿凌乱了。
　　（1）把put操作添加到writeAsyncBuffer队列里面，符合条件（自动flush或者超过了阀值writeBufferSize）就通过AsyncProcess异步批量提交。
　　（2）在提交之前，我们要根据每个rowkey找到它们归属的region server，这个定位的过程是通过HConnection的locateRegion方法获得的，然后再把这些rowkey按照HRegionLocation分组。
　　（3）通过多线程，一个HRegionLocation构造MultiServerCallable<Row>，然后通过rpcCallerFactory.<MultiResponse> newCaller()执行调用，忽略掉失败重新提交和错误处理，客户端的提交操作到此结束。

2.Delete操作
对于Delete，我们也可以通过以下代码执行一个delete操作

  1. Delete del = new Delete(rowkey);

  2. table.delete(del);
复制代码

这个操作比较干脆，new一个RegionServerCallable<Boolean>,直接走rpc了，爽快啊。
  1. RegionServerCallable<Boolean> callable = new RegionServerCallable<Boolean>(connection,

  2.         tableName, delete.getRow()) {

  3.       public Boolean call() throws IOException {

  4.         try {

  5.           MutateRequest request = RequestConverter.buildMutateRequest(

  6.             getLocation().getRegionInfo().getRegionName(), delete);

  7.           MutateResponse response = getStub().mutate(null, request);

  8.           return Boolean.valueOf(response.getProcessed());

  9.         } catch (ServiceException se) {

  10.           throw ProtobufUtil.getRemoteException(se);

  11.         }

  12.       }

  13.     };

  14. rpcCallerFactory.<Boolean> newCaller().callWithRetries(callable, this.operationTimeout);
复制代码


这里面注意一下这行MutateResponse response = getStub().mutate(null, request);
　　getStub()返回的是一个ClientService.BlockingInterface接口，实现这个接口的类是HRegionServer，这样子我们就知道它在服务端执行了HRegionServer里面的mutate方法。

3.Get操作
　　get操作也和delete一样简单

  1. Get get = new Get(rowkey);

  2. Result row = table.get(get);
复制代码

get操作也没几行代码，还是直接走的rpc
  1. public Result get(final Get get) throws IOException {

  2.     RegionServerCallable<Result> callable = new RegionServerCallable<Result>(this.connection,

  3.         getName(), get.getRow()) {

  4.       public Result call() throws IOException {

  5.         return ProtobufUtil.get(getStub(), getLocation().getRegionInfo().getRegionName(), get);

  6.       }

  7.     };

  8.     return rpcCallerFactory.<Result> newCaller().callWithRetries(callable, this.operationTimeout);

  9. }
复制代码

注意里面的ProtobufUtil.get操作，它其实是构建了一个GetRequest，需要的参数是regionName和get，然后走HRegionServer的get方法，返回一个GetResponse
  1. public static Result get(final ClientService.BlockingInterface client,

  2.       final byte[] regionName, final Get get) throws IOException {

  3.     GetRequest request =

  4.       RequestConverter.buildGetRequest(regionName, get);

  5.     try {

  6.       GetResponse response = client.get(null, request);

  7.       if (response == null) return null;

  8.       return toResult(response.getResult());

  9.     } catch (ServiceException se) {

  10.       throw getRemoteException(se);

  11.     }

  12. }
复制代码

4.批量操作

　　针对put、delete、get都有相应的操作的方式：
　　1.Put(list)操作，很多童鞋以为这个可以提高写入速度，其实无效。。。为啥？因为你构造了一个list进去，它再遍历一下list，执行doPut操作。。。。反而还慢点。
　　2.delete和get的批量操作走的都是connection.processBatchCallback(actions, tableName, pool, results, callback)，具体的实现在HConnectionManager的静态类HConnectionImplementation里面，结果我们惊人的发现：

  1. AsyncProcess<?> asyncProcess = createAsyncProcess(tableName, pool, cb, conf);

  2. asyncProcess.submitAll(list);

  3. asyncProcess.waitUntilDone();
复制代码

它走的还是put一样的操作，既然是一样的，何苦代码写得那么绕呢？

5.查询操作
　　现在讲一下scan吧，这个操作相对复杂点。还是老规矩，先上一下代码吧。

  1. Scan scan = new Scan();

  2.         //scan.setTimeRange(new Date("20140101").getTime(), new Date("20140429").getTime());

  3.         scan.setBatch(10);

  4.         scan.setCaching(10);

  5.         scan.setStartRow(Bytes.toBytes("cenyuhai-00000-20140101"));

  6.         scan.setStopRow(Bytes.toBytes("cenyuhai-zzzzz-201400429"));

  7.         //如果设置为READ_COMMITTED，它会取当前的时间作为读的检查点，在这个时间点之后的就排除掉了

  8.         scan.setIsolationLevel(IsolationLevel.READ_COMMITTED);

  9.         RowFilter rowFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("pattern"));

  10.         ResultScanner resultScanner = table.getScanner(scan);

  11.         Result result = null;

  12.         while ((result = resultScanner.next()) != null) {

  13.             //自己处理去吧...

  14.         }
复制代码

这个是带正则表达式的模糊查询的scan查询，Scan这个类是包括我们查询所有需要的参数，batch和caching的设置，在我的另外一篇文章里面有写《hbase客户端设置缓存优化查询》。

Scan查询的时候，设置StartRow和StopRow可是重头戏，假设我这里要查我01月01日到04月29日总共发了多少业务，中间是业务类型，但是我可能是所有的都查，或者只查一部分，在所有都查的情况下，我就不能设置了，那但是StartRow和StopRow我不能空着啊，所以这里可以填00000-zzzzz，只要保证它在这个区间就可以了，然后我们加了一个RowFilter，然后引入了正则表达式，之前好多人一直在问啊问的，不过我这个例子，其实不要也可以，因为是查所有业务的，在StartRow和StopRow之间的都可以要。

　　好的，我们接着看，F3进入getScanner方法

  1. if (scan.isSmall()) {

  2.       return new ClientSmallScanner(getConfiguration(), scan, getName(), this.connection);

  3. }

  4. return new ClientScanner(getConfiguration(), scan, getName(), this.connection);
复制代码

这个scan还分大小, 没关系，我们进入ClientScanner看一下吧， 在ClientScanner的构造方法里面发现它会去调用nextScanner去初始化一个ScannerCallable。好的，我们接着来到ScannerCallable里面，这里需要注意的是它的两个方法，prepare和call方法。在prepare里面它主要干了两个事情，获得region的HRegionLocation和ClientService.BlockingInterface接口的实例，之前说过这个继承这个接口的只有Region Server的实现类。
  1.  public void prepare(final boolean reload) throws IOException {

  2.     this.location = connection.getRegionLocation(tableName, row, reload); 　　 //HConnection.getClient()这个方法简直就是神器啊

  3.     setStub(getConnection().getClient(getLocation().getServerName()));

  4.   }
复制代码

ok，我们下面看看call方法吧
  1. public Result [] call() throws IOException {

  2.     　// 第一次走的地方，开启scanner

  3.       if (scannerId == -1L) {

  4.         this.scannerId = openScanner();

  5.       } else {

  6.         Result [] rrs = null;

  7.         ScanRequest request = null;

  8.         try {

  9.           request = RequestConverter.buildScanRequest(scannerId, caching, false, nextCallSeq);

  10.           ScanResponse response = null; 　　　　　　

  11. 　　　　　　// 准备用controller去携带返回的数据，这样的话就不用进行protobuf的序列化了           

  12. 　　　　　　PayloadCarryingRpcController controller = new PayloadCarryingRpcController();          

  13. 　　　　　　controller.setPriority(getTableName());

  14.           response = getStub().scan(controller, request);

  15.           nextCallSeq++;

  16.           long timestamp = System.currentTimeMillis();

  17.           // Results are returned via controller

  18.           CellScanner cellScanner = controller.cellScanner();

  19.           rrs = ResponseConverter.getResults(cellScanner, response);

  20. 　　　   } catch (IOException e) { 　　　　　　 　　　　　　

  21.         } 　　　　

  22. 　　　 } 　　　　return rrs;

  23.      

  24.     }

  25.     return null;

  26.   }
复制代码


在call方法里面，我们可以看得出来，实例化ScanRequest，然后调用scan方法的时候把PayloadCarryingRpcController传过去，这里跟踪了一下，如果设置了codec的就从PayloadCarryingRpcController里面返回结果，否则从response里面返回。

好的，下面看next方法吧。

  1. @Override

  2.     public Result next() throws IOException { if (cache.size() == 0) {

  3.         Result [] values = null;

  4.         long remainingResultSize = maxScannerResultSize;

  5.         int countdown = this.caching; 　　　　 

  6. 　　　　 // 设置获取数据的条数         

  7. 　　　　 callable.setCaching(this.caching);

  8.         boolean skipFirst = false;

  9.         boolean retryAfterOutOfOrderException  = true;

  10.         do {

  11.   　　　　　　if (skipFirst) {

  12. 　　　　　　　　 // 上次读的最后一个，这次就不读了，直接跳过就是了

  13.               callable.setCaching(1);

  14.               values = this.caller.callWithRetries(callable);

  15.               callable.setCaching(this.caching);

  16.               skipFirst = false;

  17.             }

  18. 　　　　　　　values = this.caller.callWithRetries(callable);

  19.        　　 if (values != null && values.length > 0) {

  20.             for (Result rs : values) { 　　　　　　　　 //缓存起来               cache.add(rs);

  21.               for (Cell kv : rs.rawCells()) {//计算出keyvalue的大小，然后减去

  22.                 remainingResultSize -= KeyValueUtil.ensureKeyValue(kv).heapSize();

  23.               }

  24.               countdown--;

  25.               this.lastResult = rs;

  26.             }

  27.            }

  28.           // Values == null means server-side filter has determined we must STOP

  29.         } while (remainingResultSize > 0 && countdown > 0 && nextScanner(countdown, values == null));

  30.       　

  31. 　　　　 //缓存里面有就从缓存里面取       

  32. 　　　　 if (cache.size() > 0) {

  33.           return cache.poll();

  34.         }

  35. 　　　　 return null;

  36.     }
复制代码


从next方法里面可以看出来，它是一次取caching条数据，然后下一次获取的时候，先把上次获取的最后一个给排除掉，再获取下来保存在cache当中，只要缓存不空，就一直在缓存里面取。
　　好了，至此Scan到此结束。






hbase源码系列（二）HTable 探秘hbase源码系列（三）Client如何找到正确的Region Server
hbase源码系列（四）数据模型-表定义和列族定义的具体含义
hbase源码系列（五）Trie单词查找树
hbase源码系列（六）HMaster启动过程
hbase源码系列（七）Snapshot的过程
hbase源码系列（八）从Snapshot恢复表
hbase源码系列（九）StoreFile存储格式
hbase源码系列（十）HLog与日志恢复
hbase源码系列（十一）Put、Delete在服务端是如何处理？
hbase源码系列（十二）Get、Scan在服务端是如何处理？
hbase源码系列（十三）缓存机制MemStore与Block Cache
hbase源码之如何查询出来下一个KeyValue
hbase源码Compact和Split





欢迎加入about云群90371779 、322273151、432264021 ，云计算爱好者群，亦可关注about云腾讯认证空间||关注本站微信
回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
坎蒂丝_Swan

高级会员

关注

  • 148

    主题

  • 148

    帖子

  • 52

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000