hbase源码系列(二)HTable 如何访问客户端

时间:2014-04-27 17:34:53   收藏:0   阅读:790

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

  现在我们讲一下HTable吧,为什么讲HTable,因为这是我们最常见的一个类,这是我们对hbase中数据的操作的入口。

  

1.Put操作

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

mamicode.com,码迷
HBaseConfiguration conf =  (HBaseConfiguration) HBaseConfiguration.create();
byte[] rowkey = Bytes.toBytes("cenyuhai");
byte[] family = Bytes.toBytes("f");
byte[] qualifier = Bytes.toBytes("name");
byte[] value = Bytes.toBytes("岑玉海");
        
HTable table = new HTable(conf, "test");
Put put = new Put(rowkey);
put.add(family,qualifier,value);
        
table.put(put);
mamicode.com,码迷

 

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

mamicode.com,码迷
  /** 实际提交数据所用的类 */
protected
HConnection connection;/** 需要提交的数据的列表 */ protected List<Row> writeAsyncBuffer = new LinkedList<Row>();
/** flush的size */
private long writeBufferSize; /** 是否自动flush */ private boolean autoFlush; /** 当前的数据的size,达到指定的size就要提交 */ protected long currentWriteBufferSize; protected int scannerCaching; private int maxKeyValueSize; private ExecutorService pool; // For Multi
/** 异步提交 */ protected AsyncProcess<Object> ap;
** rpc工厂 */
private RpcRetryingCallerFactory rpcCallerFactory;
mamicode.com,码迷

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

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

    doPut(put);
    if (autoFlush) {
      flushCommits();
    }

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

mamicode.com,码迷
if (ap.hasError()){
      backgroundFlushCommits(true);
}
currentWriteBufferSize += put.heapSize();
writeAsyncBuffer.add(put);
while (currentWriteBufferSize > writeBufferSize) {
    backgroundFlushCommits(false);
}
mamicode.com,码迷

  写下来,让我们看看backgroundFlushCommits这个方法吧,它的核心就这么一句ap.submit(writeAsyncBuffer, true) ,如果出错了的话,就报错了。所以网上所有关于客户端调优的方法里面无非就这么几种:

1)关闭autoFlush

2)关闭wal日志

3)把writeBufferSize设大一点,一般说是设置成5MB

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

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

  

mamicode.com,码迷
      int posInList = -1;
      Iterator<? extends Row> it = rows.iterator();
      while (it.hasNext()) {
        Row r = it.next();
        //为row定位
        HRegionLocation loc = findDestLocation(r, 1, posInList);

        if (loc != null && canTakeOperation(loc, regionIncluded, serverIncluded)) {
          // loc is null if there is an error such as meta not available.
          Action<Row> action = new Action<Row>(r, ++posInList);
          retainedActions.add(action);
          addAction(loc, action, actionsByServer);
          it.remove();
        }
      }
mamicode.com,码迷

  循环遍历r,为每个r找到它的位置loc,loc是HRegionLocation,里面记录着这行记录所在的目标region所在的位置,loc怎么获得呢,走进findDestLocation方法里面,看到了这么一句。

  

loc = hConnection.locateRegion(this.tableName, row.getRow());

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

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

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

MultiServerCallable<Row> callable = createCallable(loc, multiAction);
......
res = createCaller(callable).callWithoutRetries(callable);

  再深挖一点,把它们的实现都扒出来吧。

mamicode.com,码迷
  protected MultiServerCallable<Row> createCallable(final HRegionLocation location,
      final MultiAction<Row> multi) {
    return new MultiServerCallable<Row>(hConnection, tableName, location, multi);
  }

  protected RpcRetryingCaller<MultiResponse> createCaller(MultiServerCallable<Row> callable) {
    return rpcCallerFactory.<MultiResponse> newCaller();
  }
mamicode.com,码迷

  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操作

Delete del = new Delete(rowkey);
table.delete(del);

  这个操作比较干脆,new一个RegionServerCallable<Boolean>,直接走rpc了,爽快啊。

mamicode.com,码迷
RegionServerCallable<Boolean> callable = new RegionServerCallable<Boolean>(connection,
        tableName, delete.getRow()) {
      public Boolean call() throws IOException {
        try {
          MutateRequest request = RequestConverter.buildMutateRequest(
            getLocation().getRegionInfo().getRegionName(), delete);
          MutateResponse response = getStub().mutate(null, request);
          return Boolean.valueOf(response.getProcessed());
        } catch (ServiceException se) {
          throw ProtobufUtil.getRemoteException(se);
        }
      }
    };
rpcCallerFactory.<Boolean> newCaller().callWithRetries(callable, this.operationTimeout);
mamicode.com,码迷

 这里面注意一下这行MutateResponse response = getStub().mutate(null, request);

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

3.Get操作

  get操作也和delete一样简单

  

Get get = new Get(rowkey);
Result row = table.get(get);

  get操作也没几行代码,还是直接走的rpc

mamicode.com,码迷
public Result get(final Get get) throws IOException {
    RegionServerCallable<Result> callable = new RegionServerCallable<Result>(this.connection,
        getName(), get.getRow()) {
      public Result call() throws IOException {
        return ProtobufUtil.get(getStub(), getLocation().getRegionInfo().getRegionName(), get);
      }
    };
    return rpcCallerFactory.<Result> newCaller().callWithRetries(callable, this.operationTimeout);
}
mamicode.com,码迷

  注意里面的ProtobufUtil.get操作,它其实是构建了一个GetRequest,需要的参数是regionName和get,然后走HRegionServer的get方法,返回一个GetResponse

mamicode.com,码迷
public static Result get(final ClientService.BlockingInterface client,
      final byte[] regionName, final Get get) throws IOException {
    GetRequest request =
      RequestConverter.buildGetRequest(regionName, get);
    try {
      GetResponse response = client.get(null, request);
      if (response == null) return null;
      return toResult(response.getResult());
    } catch (ServiceException se) {
      throw getRemoteException(se);
    }
}
mamicode.com,码迷

 

 4.批量操作

  

  针对put、delete、get都有相应的操作的方式:

  1.Put(list)操作,很多童鞋以为这个可以提高写入速度,其实无效。。。为啥?因为你构造了一个list进去,它再遍历一下list,执行doPut操作。。。。反而还慢点。

  2.delete和get的批量操作走的都是connection.processBatchCallback(actions, tableName, pool, results, callback),具体的实现在HConnectionManager的静态类HConnectionImplementation里面,结果我们惊人的发现:

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

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

  okay,HTable到现在就告一段落了,后续会有在Region Server这端关于put、delete、get的处理的文章,敬请期待。

 

 

 

 

 

 

  

 

hbase源码系列(二)HTable 如何访问客户端,码迷,mamicode.com

评论(0
© 2014 mamicode.com 版权所有 京ICP备13008772号-2  联系我们:gaon5@hotmail.com
迷上了代码!