HBase BlockCache

2022-11-06 10:38:42

1. Cache 读写 

调用逻辑： 

hmaster.handleCreateTable->HRegion.createHRegion-> HRegion. initialize->initializeRegionInternals->instantiateHStore 

->Store.Store->new CacheConfig(conf, family)-> CacheConfig.instantiateBlockCache->new LruBlockCache 

传入参数

  1. /**
  2. * Configurable constructor.  Use this constructor if not using defaults.
  3. * @param maxSize maximum size of this cache, in bytes
  4. * @param blockSize expected average size of blocks, in bytes
  5. * @param evictionThread whether to run evictions in a bg thread or not
  6. * @param mapInitialSize initial size of backing ConcurrentHashMap
  7. * @param mapLoadFactor initial load factor of backing ConcurrentHashMap
  8. * @param mapConcurrencyLevel initial concurrency factor for backing CHM
  9. * @param minFactor percentage of total size that eviction will evict until
  10. * @param acceptableFactor percentage of total size that triggers eviction
  11. * @param singleFactor percentage of total size for single-access blocks
  12. * @param multiFactor percentage of total size for multiple-access blocks
  13. * @param memoryFactor percentage of total size for in-memory blocks
  14. */
  15. public LruBlockCache(long maxSize, long blockSize, boolean evictionThread,
  16. int mapInitialSize, float mapLoadFactor, int mapConcurrencyLevel,
  17. float minFactor, float acceptableFactor,
  18. float singleFactor, float multiFactor, float memoryFactor)

new LruBlockCache时除了设置默认的参数外，还会创建evictionThread并wait和一个定时打印的线程StatisticsThread

当执行HFileReaderV2的readBlock时，会先看判断是否开户了Cache ，如果开启，则使用cache中block

  1. // Check cache for block. If found return.
  2. if (cacheConf.isBlockCacheEnabled()) {
  3. // Try and get the block from the block cache.  If the useLock variable is true then this
  4. // is the second time through the loop and it should not be counted as a block cache miss.
  5. HFileBlock cachedBlock = (HFileBlock)
  6. cacheConf.getBlockCache().getBlock(cacheKey, cacheBlock, useLock);
  7. if (cachedBlock != null) {
  8. BlockCategory blockCategory =
  9. cachedBlock.getBlockType().getCategory();
  10. getSchemaMetrics().updateOnCacheHit(blockCategory, isCompaction);
  11. if (cachedBlock.getBlockType() == BlockType.DATA) {
  12. HFile.dataBlockReadCnt.incrementAndGet();
  13. }
  14. validateBlockType(cachedBlock, expectedBlockType);
  15. // Validate encoding type for encoded blocks. We include encoding
  16. // type in the cache key, and we expect it to match on a cache hit.
  17. if (cachedBlock.getBlockType() == BlockType.ENCODED_DATA &&
  18. cachedBlock.getDataBlockEncoding() !=
  19. dataBlockEncoder.getEncodingInCache()) {
  20. throw new IOException(“Cached block under key ” + cacheKey + “ ” +
  21. “has wrong encoding: ” + cachedBlock.getDataBlockEncoding() +
  22. “ (expected: ” + dataBlockEncoder.getEncodingInCache() + “)”);
  23. }
  24. return cachedBlock;
  25. }
  26. // Carry on, please load.
  27. }

在getBlock方法中，会更新一些统计数据，重要的时更新

  1. BlockPriority.SINGLE为BlockPriority.MULTI
  2. public Cacheable getBlock(BlockCacheKey cacheKey, boolean caching, boolean repeat) {
  3. CachedBlock cb = map.get(cacheKey);
  4. if(cb == null) {
  5. if (!repeat) stats.miss(caching);
  6. return null;
  7. }
  8. stats.hit(caching);
  9. cb.access(count.incrementAndGet());
  10. return cb.getBuffer();
  11. }

——————— 

若是第一次读，则将block加入Cache.

  1. // Cache the block if necessary
  2. if (cacheBlock && cacheConf.shouldCacheBlockOnRead(
  3. hfileBlock.getBlockType().getCategory())) {
  4. cacheConf.getBlockCache().cacheBlock(cacheKey, hfileBlock,
  5. cacheConf.isInMemory());
  6. }

2. LRU evict

写入cache时就是将block加入到 一个 ConcurrentHashMap中，并更新Metrics,之后判断if(newSize > acceptableSize() && !evictionInProgress), acceptableSize是初始化时给的值(long)Math.floor(this.maxSize
* this.acceptableFactor)，acceptableFactor是一个百分比，是可以配置的：”hbase.lru.blockcache.acceptable.factor”(0.85f)， 这里的意思就是判断总Size是不是大于这个值，如果大于并且没有正在执行的eviction线程，
那么就执行evict。

  1. /**
  2. * Cache the block with the specified name and buffer.
  3. * <p>
  4. * It is assumed this will NEVER be called on an already cached block.  If
  5. * that is done, an exception will be thrown.
  6. * @param cacheKey block’s cache key
  7. * @param buf block buffer
  8. * @param inMemory if block is in-memory
  9. */
  10. public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory) {
  11. CachedBlock cb = map.get(cacheKey);
  12. if(cb != null) {
  13. throw new RuntimeException(“Cached an already cached block”);
  14. }
  15. cb = new CachedBlock(cacheKey, buf, count.incrementAndGet(), inMemory);
  16. long newSize = updateSizeMetrics(cb, false);
  17. map.put(cacheKey, cb);
  18. elements.incrementAndGet();
  19. if(newSize > acceptableSize() && !evictionInProgress) {
  20. runEviction();
  21. }
  22. }

在evict方法中， 

1. 计算总size和需要free的size, minsize = (long)Math.floor(this.maxSize * this.minFactor);其中minFactor是可配置的”hbase.lru.blockcache.min.factor”（0.75f）;

  1. long currentSize = this.size.get();
  2. long bytesToFree = currentSize - minSize();

2. 初始化三种BlockBucket：bucketSingle，bucketMulti，bucketMemory并遍历map,按照三种类型分别add进各自的queue（MinMaxPriorityQueue.expectedSize(initialSize).create();）中， 并按照访问的次数逆序。 

三种类型的区别是： 

    SINGLE对应第一次读的 

    MULTI对应多次读 

    MEMORY是设定column family中的IN_MEMORY为true的

  1. // Instantiate priority buckets
  2. BlockBucket bucketSingle = new BlockBucket(bytesToFree, blockSize,
  3. singleSize());
  4. BlockBucket bucketMulti = new BlockBucket(bytesToFree, blockSize,
  5. multiSize());
  6. BlockBucket bucketMemory = new BlockBucket(bytesToFree, blockSize,
  7. memorySize());

其中三种BlockBuckt Size大小分配比例默认是： 

  static final float DEFAULT_SINGLE_FACTOR = 0.25f; 

  static final float DEFAULT_MULTI_FACTOR = 0.50f; 

  static final float DEFAULT_MEMORY_FACTOR = 0.25f;

  1. private long singleSize() {
  2. return (long)Math.floor(this.maxSize * this.singleFactor * this.minFactor);
  3. }
  4. private long multiSize() {
  5. return (long)Math.floor(this.maxSize * this.multiFactor * this.minFactor);
  6. }
  7. private long memorySize() {
  8. return (long)Math.floor(this.maxSize * this.memoryFactor * this.minFactor);
  9. }

并将三种BlockBuckt 加入到优先队列中，按照totalSize – bucketSize排序，,再计算需要free大小，执行free：

  1. PriorityQueue<BlockBucket> bucketQueue =
  2. new PriorityQueue<BlockBucket>(3);
  3. bucketQueue.add(bucketSingle);
  4. bucketQueue.add(bucketMulti);
  5. bucketQueue.add(bucketMemory);
  6. int remainingBuckets = 3;
  7. long bytesFreed = 0;
  8. BlockBucket bucket;
  9. while((bucket = bucketQueue.poll()) != null) {
  10. long overflow = bucket.overflow();
  11. if(overflow > 0) {
  12. long bucketBytesToFree = Math.min(overflow,
  13. (bytesToFree - bytesFreed) / remainingBuckets);
  14. bytesFreed += bucket.free(bucketBytesToFree);
  15. }
  16. remainingBuckets–;
  17. }

free方法中一个一个取出queue中block，由于是按照访问次数逆序，所以从后面取出就是先取出访问次数少的，将其在map中一个一个remove， 并更新Mertrics.

  1. public long free(long toFree) {
  2. CachedBlock cb;
  3. long freedBytes = 0;
  4. while ((cb = queue.pollLast()) != null) {
  5. freedBytes += evictBlock(cb);
  6. if (freedBytes >= toFree) {
  7. return freedBytes;
  8. }
  9. }
  10. return freedBytes;
  11. }
  12. otected long evictBlock(CachedBlock block) {
  13. map.remove(block.getCacheKey());
  14. updateSizeMetrics(block, true);
  15. elements.decrementAndGet();
  16. stats.evicted();
  17. return block.heapSize();

3. HBase LruBlockCache的特点是针对不同的访问次数使用不同的策略，避免频繁的更新的Cache（便如Scan）,这样更加有利于提高读的性能。

上一篇：[.NET] 使用VALIDATIONCONTEXT快速进行模型资料的验证 》简单xml创建-json转xml


下一篇：PMP备考-第二章-项目运行环境与项目经理