分享

hbase HTable之Put、delete、get等源码分析

问题导读:
1.把put操作添加到writeAsyncBuffer队列里面,符合条件如何处理?
2.Delete操作内部是如何实现的?
3.get操作使用的什么通信协议?
4.Scan查询的时候,设置StartRow和StopRow的作用是什么





现在我们讲一下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.   protected AsyncProcess<Object> ap;
  15.   ** rpc工厂 */
  16.   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.         if (loc != null && canTakeOperation(loc, regionIncluded, serverIncluded)) {
  8.           // loc is null if there is an error such as meta not available.
  9.           Action<Row> action = new Action<Row>(r, ++posInList);
  10.           retainedActions.add(action);
  11.           addAction(loc, action, actionsByServer);
  12.           it.remove();
  13.         }
  14.       }
复制代码
循环遍历r,为每个r找到它的位置loc,loc是HRegionLocation,里面记录着这行记录所在的目标region所在的位置,loc怎么获得呢,走进findDestLocation方法里面,看到了这么一句。
  1. loc = hConnection.locateRegion(this.tableName, row.getRow());
复制代码
通过表名和rowkey,使用HConnection就可以定位到它的位置,这里就先不讲定位了,稍后放一节出来讲,否则篇幅太长了,这里我们只需要记住,提交操作,是要知道它对应的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.   protected RpcRetryingCaller<MultiResponse> createCaller(MultiServerCallable<Row> callable) {
  6.     return rpcCallerFactory.<MultiResponse> newCaller();
  7.   }
复制代码
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客户端设置缓存优化查询
我们在用hbase的api对hbase进行scan操作的时候,可以设置caching和batch来提交查询效率,那它们之间的关系是啥样的呢,我们又应该如何去设置?
首先是我们的客户端代码。
1.png
当caching和batch都为1的时候,我们要返回10行具有20列的记录,就要进行201次RPC,因为每一列都作为一个单独的Result来返回,这样是我们不可以接受的。
2.png
下面展示的是当batch=3,caching=6时候的图,是一次RPCs的传递的数据。
3.png
接着我们继续看下图
4.png
一次查询20条记录的话,只需要3次RPCs,列数在10列以内的数据,取20条,20/10即可,为什么是3呢,因为还有一次RPC是用来确认的。
有个公式RPCs = (Rows * Cols per Row) / Min(Cols per Row, Batch Size)/ Scanner Caching 。
这就好说啦,这样我们就可以用来优化我们的scan查询了,在查询的时候,按照查询的列数动态设置batch,如果全查,则根据自己所有的表的大小设置一个折中的数值,caching就和分页的值一样就行。


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;       // 准备用controller去携带返回的数据,这样的话就不用进行protobuf的序列化了           
  11.       PayloadCarryingRpcController controller = new PayloadCarryingRpcController();         
  12.       controller.setPriority(getTableName());
  13.           response = getStub().scan(controller, request);
  14.           nextCallSeq++;
  15.           long timestamp = System.currentTimeMillis();
  16.           // Results are returned via controller
  17.           CellScanner cellScanner = controller.cellScanner();
  18.           rrs = ResponseConverter.getResults(cellScanner, response);
  19.       } catch (IOException e) {              
  20.         }     
  21.     }     return rrs;
  22.      
  23.     }
  24.     return null;
  25.   }
复制代码
在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;      // 设置获取数据的条数         callable.setCaching(this.caching);
  6.         boolean skipFirst = false;
  7.         boolean retryAfterOutOfOrderException  = true;
  8.         do {
  9.         if (skipFirst) {
  10.          // 上次读的最后一个,这次就不读了,直接跳过就是了
  11.               callable.setCaching(1);
  12.               values = this.caller.callWithRetries(callable);
  13.               callable.setCaching(this.caching);
  14.               skipFirst = false;
  15.             }
  16.        values = this.caller.callWithRetries(callable);
  17.           if (values != null && values.length > 0) {
  18.             for (Result rs : values) {          //缓存起来               cache.add(rs);
  19.               for (Cell kv : rs.rawCells()) {//计算出keyvalue的大小,然后减去
  20.                 remainingResultSize -= KeyValueUtil.ensureKeyValue(kv).heapSize();
  21.               }
  22.               countdown--;
  23.               this.lastResult = rs;
  24.             }
  25.            }
  26.           // Values == null means server-side filter has determined we must STOP
  27.         } while (remainingResultSize > 0 && countdown > 0 && nextScanner(countdown, values == null));
  28.        
  29.      //缓存里面有就从缓存里面取      
  30.      if (cache.size() > 0) {
  31.           return cache.poll();
  32.         }
  33.      return null;
  34.     }
复制代码

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

  好了,至此Scan到此结束。








已有(3)人评论

跳转到指定楼层
pengsuyun 发表于 2015-1-3 11:47:27
回复

使用道具 举报

Mr.k 发表于 2017-4-20 10:02:36
写的不错,赞
回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条