图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Hbase › 代码分析› hbase源码Put、Delete在服务端如何处理？
分享

hbase源码Put、Delete在服务端如何处理？

xioaxu790 2014-10-17 13:15:00 发表于 代码分析 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 0 8642
About云VIP帮助找工作和提升
本帖最后由 pig2 于 2014-10-18 00:42 编辑
问题导读
1、Put在Region Server中做了什么？
2、我们常用的方式是什么？
3、什么情况下执行prepareDeleteTimestamps方法？





在讲完之后HFile和HLog之后，今天我想分享是Put在Region Server经历些了什么？相信前面看了《HTable探秘》的朋友都会有印象，没看过的建议回去先看看，Put是通过MultiServerCallable来提交的多个Put，好，我们就先去这个类吧，在call方法里面，我们找到了这句。

responseProto = getStub().multi(controller, requestProto);
它调用了Region Server的multi方法。好，我们立即杀到HRegionServer去，搜索找到multi这个方法。
  1. public MultiResponse multi(final RpcController rpcc, final MultiRequest request)

  2.   throws ServiceException {

  3.     // RpcController是属于后门的，这样返回的数据就不用序列化了

  4.     PayloadCarryingRpcController controller = (PayloadCarryingRpcController)rpcc;

  5.     CellScanner cellScanner = controller != null? controller.cellScanner(): null;

  6.     if (controller != null) controller.setCellScanner(null);

  7.     List<CellScannable> cellsToReturn = null;

  8.      MultiResponse.Builder responseBuilder = MultiResponse.newBuilder();

  9.      //取出来所有的Action

  10.      for (RegionAction regionAction : request.getRegionActionList()) {

  11.        this.requestCount.add(regionAction.getActionCount());

  12.        RegionActionResult.Builder regionActionResultBuilder = RegionActionResult.newBuilder();

  13.        HRegion region;

  14.        try {

  15.          //获取对应的HRegion

  16.          region = getRegion(regionAction.getRegion());

  17.        } catch (IOException e) {

  18.          responseBuilder.addRegionActionResult(regionActionResultBuilder.build());

  19.          continue;  // 报告这个action有错

  20.        }

  21. 

  22.        if (regionAction.hasAtomic() && regionAction.getAtomic()) {

  23.          try {

  24.           //如果是原子操作，就走原子操作的通道

  25.            mutateRows(region, regionAction.getActionList(), cellScanner);

  26.          } catch (IOException e) {

  27.            regionActionResultBuilder.setException(ResponseConverter.buildException(e));

  28.          }

  29.        } else {

  30.          // 非原子性提交，把错误内部处理了

  31.          cellsToReturn = doNonAtomicRegionMutation(region, regionAction, cellScanner,

  32.              regionActionResultBuilder, cellsToReturn);

  33.        }

  34.        responseBuilder.addRegionActionResult(regionActionResultBuilder.build());

  35.      }

  36.      // 如果需要返回数据的话，就new一个createCellScanner扔回去

  37.      if (cellsToReturn != null && !cellsToReturn.isEmpty() && controller != null) {

  38.        controller.setCellScanner(CellUtil.createCellScanner(cellsToReturn));

  39.      }

  40.      return responseBuilder.build();

  41.    }
复制代码


这个方法里面还包括了PayloadCarryingRpcController和CellScanner可以看得出来它不只是被Put来用的，但是这些我们不管我们只看Put如何处理就行了。

1、取出来所有的action（Put），这里主要是put，因为我们调用客户端就是这么调用的，其实别的类型也可以支持，获取他们对应的region。

2、根据action的原子性来判断走哪个方法，原子性操作走mutateRows，非原子性操作走doNonAtomicRegionMutation方法，我查了一下这个Atomic到底是怎么回事，我搜索了一下代码，发现在调用HTable的mutateRow方法的时候，它设置了Atomic为true，这个是应该是支持一行数据的原子性的，有这个需求的童鞋可以尝试用这个方法，也是可以提交多个，包括Put、Delete操作。
  1. regionMutationBuilder.setAtomic(true);

  2. getStub().multi(null, request);
复制代码


我们先看doNonAtomicRegionMutation，这是我们常用的方式。
  1. 　　　List<ClientProtos.Action> mutations = null;

  2.      for (ClientProtos.Action action: actions.getActionList()) {

  3.        ClientProtos.ResultOrException.Builder resultOrExceptionBuilder = null;

  4.        try {

  5.          Result r = null;

  6.          if (action.hasGet()) {

  7.            Get get = ProtobufUtil.toGet(action.getGet());

  8.            r = region.get(get);

  9.          } else if (action.hasMutation()) {

  10.            MutationType type = action.getMutation().getMutateType();

  11.            if (type != MutationType.PUT && type != MutationType.DELETE && mutations != null &&

  12.                !mutations.isEmpty()) {

  13.              // 如果这个操作不是Put或者Delete的话，就一下子把前面的活都先干了？

  14.              doBatchOp(builder, region, mutations, cellScanner);

  15.              mutations.clear();

  16.            }

  17.            switch (type) {

  18.            case APPEND:

  19.              r = append(region, action.getMutation(), cellScanner);

  20.              break;

  21.            case INCREMENT:

  22.              r = increment(region, action.getMutation(), cellScanner);

  23.              break;

  24.            case PUT:

  25.            case DELETE:

  26.              // 前面的那些，我们都用得少，或者是不用，不用管它们，看这里就行

  27.              if (mutations == null) {

  28.                mutations = new ArrayList<ClientProtos.Action>(actions.getActionCount());

  29.             }

  30.              mutations.add(action);

  31.              break;

  32.            default:

  33.              throw new DoNotRetryIOException("Unsupported mutate type: " + type.name());

  34.           }

  35.         } else {

  36.            throw new HBaseIOException("Unexpected Action type");

  37.         }

  38.         if (r != null) {

  39.           ClientProtos.Result pbResult = null;

  40.           if (isClientCellBlockSupport()) {

  41.              pbResult = ProtobufUtil.toResultNoData(r);

  42.              //  

  43.              if (cellsToReturn == null) cellsToReturn = new ArrayList<CellScannable>();

  44.              cellsToReturn.add(r);

  45.           } else {

  46.             pbResult = ProtobufUtil.toResult(r);

  47.           }

  48.           //把result编译成Protobuf码，返回

  49.           resultOrExceptionBuilder =

  50.             ClientProtos.ResultOrException.newBuilder().setResult(pbResult);

  51.         }

  52.       } catch (IOException ie) {

  53.         resultOrExceptionBuilder = ResultOrException.newBuilder().

  54.           setException(ResponseConverter.buildException(ie));

  55.       }

  56.       if (resultOrExceptionBuilder != null) {

  57.         // Propagate index.

  58.         resultOrExceptionBuilder.setIndex(action.getIndex());

  59.         builder.addResultOrException(resultOrExceptionBuilder.build());

  60.       }

  61.     }

  62.     //进行批量操作 

  63.     if (mutations != null && !mutations.isEmpty()) {

  64.       doBatchOp(builder, region, mutations, cellScanner);

  65.     }

  66.     return cellsToReturn;
复制代码


这里面代码很多，也适配了很多种类型，是个大而全的方法，但是我们这里用到的只是把Put、Delete等的类型转换添加到mutations的列表里，然后走下面这个批量操作。

此外get的批量操作也是走的这个方法，里面它走的是HRegion.get的方法返回一个Result。
  1. doBatchOp(builder, region, mutations, cellScanner);
复制代码


doBatchOp里面的代码我就补贴了，老帖代码就没意思了。

1、还是得把Put、Delete给转换类型，这里的批量操作只支持全是Delete或者全是Put。

2、用HRegion.batchMutate方法来执行操作，返回OperationStatus数组，记录每个action的状态，是成功，还是失败，或者是别的状态。

在batchMutate里面首先就是检查是否是只读状态，然后检查是否是Meta Region的，是不执行MemStore检查了，因为MemStore的堆内存超过了阻塞队列的MemStore大小，就会报错误，太恶劣了。。。没catch的哦。
  1. long addedSize = doMiniBatchMutation(batchOp, isReplay);

  2. //MemStore的大小到了阀值，就要flush到文件了

  3. if (isFlushSize(newSize)) {

  4.    requestFlush();

  5. }
复制代码


doMiniBatchMutation就是我们的终极boss了，是个很长很臭的类，贴代码都不能一下子全贴。

1、实例化几个重要的类，后面具体会用到
  1. //日志，isInReplay是否支持重做，这里是false

  2. WALEdit walEdit = new WALEdit(isInReplay);

  3. //控制多版本的MemStore flush的结果，每次flush的w都是一样的，就好像同一批号的食品

  4. MultiVersionConsistencyControl.WriteEntry w = null;

  5. long txid = 0;

  6. //日志同步是否成功

  7. boolean walSyncSuccessful = false;

  8. boolean locked = false;
复制代码


2、检查Put和Delete里面的列族是否和Region持有的列族的定义相同，有时候我们在Delete的时候是不填列族的，这里它给这个缺的列族来一个KeyValue.Type.DeleteFamily，删除列族的类型。

3、给Row加锁，先计算hash值做key，如果该key没上过锁，就上一把锁，然后计算出来要写的action有多少个，记录到numReadyToWrite。

4、更新时间戳，把该action里面的所有的kv的时间戳更新为最新的时间戳，它这里也会把之前的没运行的也一起更新。

5、给该region加锁，这个时间点之后，就不允许读了，等待时间需要根据numReadyToWrite的数量来计算。
  1. //加锁，现在要上锁了，这段时间内不允许读

  2. lock(this.updatesLock.readLock(), numReadyToWrite);

  3. locked = true;

  4. 

  5. //等待时间

  6. final long waitTime = Math.min(maxBusyWaitDuration,

  7.         busyWaitDuration * Math.min(numReadyToWrite, maxBusyWaitMultiplier));

  8.       if (!lock.tryLock(waitTime, TimeUnit.MILLISECONDS)) {

  9.         throw new RegionTooBusyException(

  10.           "failed to get a lock in " + waitTime + "ms");

  11. }
复制代码


6、上锁之后，下面就是重头戏了，也就是Put、Delete等的重点。给这些写入memstore的数据创建一个批次号。

  1. //为这次添加进MemStore的数据添加一个批次号

  2. w = mvcc.beginMemstoreInsert();

  3. 

  4. //这是批次号的计算方式，nextWriteNumber就等于memstore的写的次数+1

  5. public WriteEntry beginMemstoreInsert() {

  6.     synchronized (writeQueue) {

  7.       long nextWriteNumber = ++memstoreWrite;

  8.       WriteEntry e = new WriteEntry(nextWriteNumber);

  9.       writeQueue.add(e);

  10.       return e;

  11.     }

  12. }
复制代码

7、把kv们写入到memstore当中，然后计算出来一个添加数据之后的新的MemStore的大小addedSize。

  1. //把kv们写入memstore

  2. long addedSize = 0;

  3. for (int i = firstIndex; i < lastIndexExclusive; i++) {

  4.     if (batchOp.retCodeDetails[i].getOperationStatusCode() != OperationStatusCode.NOT_RUN) {

  5.         continue;

  6.     }

  7.     addedSize += applyFamilyMapToMemstore(familyMaps[i], w);

  8. } 
复制代码

这个添加到MemStore里面也没啥神秘的，因为MemStore里面有两个kv的集合，它只是把kv添加到集合里面去，看下面的代码就知道了。
  1. private long applyFamilyMapToMemstore(Map<byte[], List<Cell>> familyMap,

  2.     MultiVersionConsistencyControl.WriteEntry localizedWriteEntry) {

  3.     long size = 0;try {for (Map.Entry<byte[], List<Cell>> e : familyMap.entrySet()) {

  4.         byte[] family = e.getKey();

  5.         List<Cell> cells = e.getValue();

  6.         //把kv添加到memstore当中

  7.         Store store = getStore(family);

  8.         for (Cell cell: cells) {

  9.           KeyValue kv = KeyValueUtil.ensureKeyValue(cell);

  10.           kv.setMvccVersion(localizedWriteEntry.getWriteNumber());

  11.           size += store.add(kv);

  12.         }

  13.       }

  14.     } 

  15. return size;

  16.    }
复制代码


注意这一句话kv.setMvccVersion(localizedWriteEntry.getWriteNumber());  后面会用到的。

8、把kv添加到日志当中，标志状态为成功，如果是用户设置了不写入日志的，它就不写入日志了。
  1.       Durability durability = Durability.USE_DEFAULT;

  2.       for (int i = firstIndex; i < lastIndexExclusive; i++) {

  3.         // 跳过状态不对的

  4.         if (batchOp.retCodeDetails[i].getOperationStatusCode()

  5.             != OperationStatusCode.NOT_RUN) {

  6.           continue;

  7.         }

  8.         //标志状态为成功

  9.         batchOp.retCodeDetails[i] = OperationStatus.SUCCESS;

  10. 

  11.         Mutation m = batchOp.operations[i];

  12.         //获取自定义的日志同步方式

  13.         Durability tmpDur = getEffectiveDurability(m.getDurability());

  14.         if (tmpDur.ordinal() > durability.ordinal()) {

  15.           durability = tmpDur;

  16.         }

  17.         if (tmpDur == Durability.SKIP_WAL) {

  18.           //记录日志的kv的大小，但是不写入到日志当中

  19.           recordMutationWithoutWal(m.getFamilyCellMap());

  20.           continue;

  21.         }

  22.         //把列族里面的kv全部添加到walEdit当中

  23.         addFamilyMapToWALEdit(familyMaps[i], walEdit);

  24.       }
复制代码

9、先异步添加日志，这里为什么是异步的，因为之前给上锁了，暂时不能读了，如果这里调用的是同步的方法，后果自己想象下。
  1. Mutation mutation = batchOp.operations[firstIndex];

  2. if (walEdit.size() > 0) {

  3.    //异步添加日志

  4.    txid = this.log.appendNoSync(this.getRegionInfo(), this.htableDescriptor.getTableName(),

  5.            walEdit, mutation.getClusterIds(), now, this.htableDescriptor);

  6. }
复制代码


10、释放之前创建的锁。
  1. //释放相关的锁

  2. if (locked) {

  3.     this.updatesLock.readLock().unlock();

  4.     locked = false;

  5. }

  6. releaseRowLocks(acquiredRowLocks);

  7. 11、同步日志。

  8. 

  9. if (walEdit.size() > 0) {

  10.     syncOrDefer(txid, durability);

  11. }

  12. walSyncSuccessful = true;
复制代码


12、结束该批次的操作。
  1. if (w != null) {

  2.    mvcc.completeMemstoreInsert(w);

  3.    w = null;

  4. }
复制代码


到这里其实就是结束了。但是如果添加到了MemStore里面了，但是日志没有同步成功呢？
  1. finally {

  2.   if (!walSyncSuccessful) {

  3.      //如果日志没有成功，

  4.      rollbackMemstore(batchOp, familyMaps, firstIndex, lastIndexExclusive);

  5.   }

  6.   ......  

  7. }
复制代码


一路跟踪代码下去，跟踪到代码在MemStore的rollback方法里面。
  1. KeyValue found = this.snapshot.get(kv);

  2. if (found != null && found.getMvccVersion() == kv.getMvccVersion()) {

  3.  　this.snapshot.remove(kv);

  4. }

  5. // 比较一下mvcc，相同就删除.

  6. found = this.kvset.get(kv);

  7. if (found != null && found.getMvccVersion() == kv.getMvccVersion()) {

  8.    removeFromKVSet(kv);

  9.    long s = heapSizeChange(kv, true);

  10.    this.size.addAndGet(-s);

  11. }
复制代码


比较了MvccVersion，发现是同一批次的，就干掉了。


过程写得比较凌乱了，把之前的总结一下吧：
1、做准备工作，实例化变量

2、检查Put和Delete里面的列族是否和Region持有的列族的定义相同。

3、给Row加锁，先计算hash值做key，如果该key没上过锁，就上一把锁，然后计算出来要写的action有多少个，记录到numReadyToWrite。

4、更新时间戳，把该action里面的所有的kv的时间戳更新为最新的时间戳，它这里也会把之前的没运行的也一起更新。

5、给该region加锁，这个时间点之后，就不允许读了，等待时间需要根据numReadyToWrite的数量来计算。

6、上锁之后，下面就是重头戏了，也就是Put、Delete等的重点。给这些写入memstore的数据创建一个批次号。

7、把kv们写入到memstore当中，然后计算出来一个添加数据之后的新的MemStore的大小addedSize。

8、把kv添加到日志当中，标志状态为成功，如果是用户设置了不写入日志的，它就不写入日志了。

9、先异步添加日志。

10、释放之前创建的锁。

11、同步日志。

12、结束该批次的操作。

Final、同步日志没成功的，最后根据批次回滚MemStore中的操作。

上面的过程适用于Put和Delete的批量操作，但是这里总感觉很好奇，就这样结束了，Put和Delete操作就没区别吗，那它怎么删除数据的？

返回在第4步更新时间戳的时候，发现了一些猫腻，Delete的情况执行了prepareDeleteTimestamps方法，看看吧。

  1. void prepareDeleteTimestamps(Map<byte[], List<Cell>> familyMap, byte[] byteNow)

  2.       throws IOException {

  3.     for (Map.Entry<byte[], List<Cell>> e : familyMap.entrySet()) {

  4.       byte[] family = e.getKey();

  5.       List<Cell> cells = e.getValue();

  6.       //列和count的映射

  7.       Map<byte[], Integer> kvCount = new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR);

  8. 

  9.       for (Cell cell: cells) {

  10.         KeyValue kv = KeyValueUtil.ensureKeyValue(cell);

  11.         // 如果是时间戳是最新的话就执行下面这些操作

  12.         if (kv.isLatestTimestamp() && kv.isDeleteType()) {

  13.           //new一个Get从Store里面去搜索

  14.          } else {

  15.           kv.updateLatestStamp(byteNow);

  16.         }

  17.       }

  18.     }

  19.   }
复制代码

看来一下代码，这里是上来先判断是否是最新的时间戳，我就回去看来一下Delete的构造函数，尼玛。。。
  1. public Delete(byte [] row) {

  2.     this(row, HConstants.LATEST_TIMESTAMP);

  3. }

  4. 

  5. public Delete(byte [] row, long timestamp) {

  6.     this(row, 0, row.length, timestamp);

  7. }
复制代码


只传了rowkey进去的，它就是最新的。。然后看了一下注释，凡是在这个时间点之前的所有版本的所有列，我们都要删除。

好吧，我们很无奈的宣布，我们只能走kv.isLatestTimestamp() && kv.isDeleteType()，下面是没放出来的代码。
  1. byte[] qual = kv.getQualifier();

  2.           if (qual == null) qual = HConstants.EMPTY_BYTE_ARRAY;

  3.           //想到相同列的每次+1

  4.           Integer count = kvCount.get(qual);

  5.           if (count == null) {

  6.             kvCount.put(qual, 1);

  7.           } else {

  8.             kvCount.put(qual, count + 1);

  9.           }

  10.           //更新之后把最新的count数量

  11.           count = kvCount.get(qual);

  12. 

  13.           Get get = new Get(kv.getRow());

  14.           get.setMaxVersions(count);

  15.           get.addColumn(family, qual);

  16.           //从store当中取出相应的result来

  17.           List<Cell> result = get(get, false);

  18. 

  19.           if (result.size() < count) {

  20.             // Nothing to delete 数量不够。。 更新最新的时间戳为现在的时间

  21.             kv.updateLatestStamp(byteNow);

  22.             continue;

  23.           }

  24.           //数量超过了也不行

  25.           if (result.size() > count) {

  26.             throw new RuntimeException("Unexpected size: " + result.size());

  27.           }

  28.           //取最后一个的时间戳

  29.           KeyValue getkv = KeyValueUtil.ensureKeyValue(result.get(count - 1));

  30.           //更新kv的时间戳为getkv的时间戳

  31.           Bytes.putBytes(kv.getBuffer(), kv.getTimestampOffset(),

  32.               getkv.getBuffer(), getkv.getTimestampOffset(), Bytes.SIZEOF_LONG);<span style="line-height: 1.5; background-color: rgb(255, 255, 255);"> </span>
复制代码


这里又干了一个Get操作，把列族的多个版本的内容取出来，如果数量不符合预期也会有问题，但是这后面操作的中心思想就是：

（a）按照预期来说，取出来的少了，就设置删除的时间戳为现在;

（b）取出来的多了，就报错;

（c）刚好的,就把Delete的时间戳设置为最大的那个的时间戳，但即便是这样也没有删除数据。

回到这里我又想起来，只有在Compaction之后，hbase的文件才会变小，难道是在那个时候删除的？那在删除之前，我们进行Get或者Scan操作的时候，会不会读到这些没有被删除的数据呢？

好，让我们拭目以待。



关注公众号，获取大数据、人工智能20套、区块链资源5阶段等资源，随时更新，获取最新技术资源
回复

使用道具 举报

自定义广告语

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

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

本版积分规则

发表新帖
xioaxu790

金牌会员

关注

  • 851

    主题

  • 1000

    帖子

  • 108

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

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

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

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