冷熱數(shù)據(jù)分離 | Alluxio元數(shù)據(jù)管理策略

點(diǎn)擊上方藍(lán)色字體，選擇“設(shè)為星標(biāo)”

本文作者：林意群

原文地址：http://suo.im/5Xcmci

一.Alluxio概述

• 最近訪問的熱點(diǎn)元數(shù)據(jù)，做內(nèi)存緩存，叫做cached layer。

• 很久沒有訪問過的數(shù)據(jù)((也可稱作冷數(shù)據(jù))，做持久化保存存，叫做persisted layer。

• 及時(shí)將那些訪問頻率降低的熱點(diǎn)數(shù)據(jù)移除并寫出到baking store里去。

• 有新的數(shù)據(jù)訪問來時(shí)，將這些數(shù)據(jù)從baking store讀出來并加載到cache里去。

When Does Cleanup Happen?Caches?built?with?CacheBuilder?do?not?perform?cleanup?and?evict?values?"automatically,"?or?instantly?after?a?value?expires,?or?anything?of?the?sort.?Instead,?it?performs?small?amounts?of?maintenance?during?write?operations,?or?during?occasional?read?operations?if?writes?are?rare.The?reason?for?this?is?as?follows:?if?we?wanted?to?perform?Cache?maintenance?continuously,?we?would?need?to?create?a?thread,?and?its?operations?would?be?competing?with?user?operations?for?shared?locks.?Additionally,?some?environments?restrict?the?creation?of?threads,?which?would?make?CacheBuilder?unusable?in?that?environment.Instead,?we?put?the?choice?in?your?hands.?If?your?cache?is?high-throughput,?then?you?don't?have?to?worry?about?performing?cache?maintenance?to?clean?up?expired?entries?and?the?like.?If?your?cache?does?writes?only?rarely?and?you?don't?want?cleanup?to?block?cache?reads,?you?may?wish?to?create?your?own?maintenance?thread?that?calls?Cache.cleanUp()?at?regular?intervals.If?you?want?to?schedule?regular?cache?maintenance?for?a?cache?which?only?rarely?has?writes,?just?schedule?the?maintenance?using?ScheduledExecutorService.

public final class CachingInodeStore implements InodeStore, Closeable {  private static final Logger LOG = LoggerFactory.getLogger(CachingInodeStore.class);  // Backing store用戶數(shù)據(jù)寫出持久化的store  private final InodeStore mBackingStore;  private final InodeLockManager mLockManager;
  // Cache recently-accessed inodes.  @VisibleForTesting  final InodeCache mInodeCache;
  // Cache recently-accessed inode tree edges.  @VisibleForTesting  final EdgeCache mEdgeCache;
  @VisibleForTesting  final ListingCache mListingCache;
  // Starts true, but becomes permanently false if we ever need to spill metadata to the backing  // store. When true, we can optimize lookups for non-existent inodes because we don't need to  // check the backing store. We can also optimize getChildren by skipping the range query on the  // backing store.  private volatile boolean mBackingStoreEmpty;  ...
  public CachingInodeStore(InodeStore backingStore, InodeLockManager lockManager) {    mBackingStore = backingStore;    mLockManager = lockManager;    AlluxioConfiguration conf = ServerConfiguration.global();    int maxSize = conf.getInt(PropertyKey.MASTER_METASTORE_INODE_CACHE_MAX_SIZE);    Preconditions.checkState(maxSize > 0,        "Maximum cache size %s must be positive, but is set to %s",        PropertyKey.MASTER_METASTORE_INODE_CACHE_MAX_SIZE.getName(), maxSize);    float highWaterMarkRatio = ConfigurationUtils.checkRatio(conf,        PropertyKey.MASTER_METASTORE_INODE_CACHE_HIGH_WATER_MARK_RATIO);    // 最高水位的計(jì)算    int highWaterMark = Math.round(maxSize * highWaterMarkRatio);    float lowWaterMarkRatio = ConfigurationUtils.checkRatio(conf,        PropertyKey.MASTER_METASTORE_INODE_CACHE_LOW_WATER_MARK_RATIO);    Preconditions.checkState(lowWaterMarkRatio <= highWaterMarkRatio,        "low water mark ratio (%s=%s) must not exceed high water mark ratio (%s=%s)",        PropertyKey.MASTER_METASTORE_INODE_CACHE_LOW_WATER_MARK_RATIO.getName(), lowWaterMarkRatio,        PropertyKey.MASTER_METASTORE_INODE_CACHE_HIGH_WATER_MARK_RATIO, highWaterMarkRatio);    // 最低水位的計(jì)算    int lowWaterMark = Math.round(maxSize * lowWaterMarkRatio);
    mBackingStoreEmpty = true;    CacheConfiguration cacheConf = CacheConfiguration.newBuilder().setMaxSize(maxSize)        .setHighWaterMark(highWaterMark).setLowWaterMark(lowWaterMark)        .setEvictBatchSize(conf.getInt(PropertyKey.MASTER_METASTORE_INODE_CACHE_EVICT_BATCH_SIZE))        .build();    // 將上述cache相關(guān)配置值傳入cache中    mInodeCache = new InodeCache(cacheConf);    mEdgeCache = new EdgeCache(cacheConf);    mListingCache = new ListingCache(cacheConf);??}

  class InodeCache extends Cache> {    public InodeCache(CacheConfiguration conf) {      super(conf, "inode-cache", MetricKey.MASTER_INODE_CACHE_SIZE);    }    ...}

public abstract class Cache implements Closeable {  private static final Logger LOG = LoggerFactory.getLogger(Cache.class);
  private final int mMaxSize;  // cache的高水位值，當(dāng)當(dāng)前cache entry總數(shù)超過此值時(shí)，會觸發(fā)entry的寫出  private final int mHighWaterMark;  // cache的低水位值，每次cache寫出清理后的entry總數(shù)  private final int mLowWaterMark;  // 每次過期寫出entry的批量大小  private final int mEvictBatchSize;  private final String mName;  // cache map，為了保證線程安全，使用了ConcurrentHashMap  @VisibleForTesting  final ConcurrentHashMap mMap;  // TODO(andrew): Support using multiple threads to speed up backing store writes.  // Thread for performing eviction to the backing store.  @VisibleForTesting  // entry移除寫出線程  final EvictionThread mEvictionThread;??...

class EvictionThread extends Thread {    @VisibleForTesting    volatile boolean mIsSleeping = true;
    // 存儲需要被清理出去的cache entry    private final List mEvictionCandidates = new ArrayList<>(mEvictBatchSize);    private final List mDirtyEvictionCandidates = new ArrayList<>(mEvictBatchSize);    private final Logger mCacheFullLogger = new SamplingLogger(LOG, 10L * Constants.SECOND_MS);
    ...
    @Override    public void run() {      while (!Thread.interrupted()) {        // 如果當(dāng)前map總entry數(shù)未超過高水位置，則線程進(jìn)行wait等待        while (!overHighWaterMark()) {          synchronized (mEvictionThread) {            if (!overHighWaterMark()) {              try {                mIsSleeping = true;                mEvictionThread.wait();                mIsSleeping = false;              } catch (InterruptedException e) {                return;              }            }          }        }        if (cacheIsFull()) {          mCacheFullLogger.warn(              "Metastore {} cache is full. Consider increasing the cache size or lowering the "                  + "high water mark. size:{} lowWaterMark:{} highWaterMark:{} maxSize:{}",              mName, mMap.size(), mLowWaterMark, mHighWaterMark, mMaxSize);        }        // 如果當(dāng)前map總entry數(shù)超過高水位置，則開始準(zhǔn)備進(jìn)行entry的寫出清理，map entry數(shù)量清理至低水位置        evictToLowWaterMark();      }    }}

    private void evictToLowWaterMark() {      long evictionStart = System.nanoTime();      // 計(jì)算此處entry移除會被移除的數(shù)量      int toEvict = mMap.size() - mLowWaterMark;      // 當(dāng)前移除entry的計(jì)數(shù)累加值      int evictionCount = 0;      // 進(jìn)行entry的寫出移除      while (evictionCount < toEvict) {        if (!mEvictionHead.hasNext()) {          mEvictionHead = mMap.values().iterator();        }        // 遍歷mapentry，進(jìn)行需要被移除的entry數(shù)的收集        fillBatch(toEvict - evictionCount);        // 進(jìn)行entry的寫出清理        evictionCount += evictBatch();      }      if (evictionCount > 0) {        LOG.debug("{}: Evicted {} entries in {}ms", mName, evictionCount,            (System.nanoTime() - evictionStart) / Constants.MS_NANO);      }????}

    private void fillBatch(int count) {      // 單次移除entry數(shù)的上限值設(shè)定      int targetSize = Math.min(count, mEvictBatchSize);      // 當(dāng)待移除entry未達(dá)到目標(biāo)值時(shí)，繼續(xù)遍歷map尋找未被引用的entry      while (mEvictionCandidates.size() < targetSize && mEvictionHead.hasNext()) {        Entry candidate = mEvictionHead.next();        // 如果entry被外界引用，則將其引用值標(biāo)記為false，下次如果還遍歷到此entry，此entry將被收集移除        // 當(dāng)entry被會訪問時(shí)，其reference值會被標(biāo)記為true。        if (candidate.mReferenced) {          candidate.mReferenced = false;          continue;        }        // 如果此entry已經(jīng)被標(biāo)記為沒有引用，則加入到待移除entry列表內(nèi)        mEvictionCandidates.add(candidate);        if (candidate.mDirty) {          mDirtyEvictionCandidates.add(candidate);        }      }    }

    private int evictBatch() {      int evicted = 0;      if (mEvictionCandidates.isEmpty()) {        return evicted;      }      // 進(jìn)行entry的寫出，entry分為兩類      // 如果entry值和baking store里保存的是一致的話：則直接從map里進(jìn)行移除即可      // 如果entry值和baking store對比是發(fā)生過更新的，則額外還需要進(jìn)行flush寫出，然后map里再進(jìn)行移除      flushEntries(mDirtyEvictionCandidates);      for (Entry entry : mEvictionCandidates) {        if (evictIfClean(entry)) {          evicted++;        }      }      mEvictionCandidates.clear();      mDirtyEvictionCandidates.clear();      return evicted;    }

• 第一類，只需從cache map中進(jìn)行移除

• 第二類，從cache map中進(jìn)行移除，還需要寫出到baking store。

  protected class Entry {    protected K mKey;    // null value means that the key has been removed from the cache, but still needs to be removed    // from the backing store.    @Nullable    protected V mValue;
    // Whether the entry is out of sync with the backing store. If mDirty is true, the entry must be    // flushed to the backing store before it can be evicted.    protected volatile boolean mDirty = true;,,,

  /**   * Attempts to flush the given entries to the backing store.   *   * The subclass is responsible for setting each candidate's mDirty field to false on success.   *   * @param candidates the candidate entries to flush   */??protected?abstract?void?flushEntries(List?candidates)

 /**   * Writes a key/value pair to the cache.   *   * @param key the key   * @param value the value   */  public void put(K key, V value) {    mMap.compute(key, (k, entry) -> {      // put操作callback接口方法      onPut(key, value);      // 如果是cache已經(jīng)滿了，則直接寫出到baking store里      if (entry == null && cacheIsFull()) {        writeToBackingStore(key, value);        return null;      }      if (entry == null || entry.mValue == null) {        onCacheUpdate(key, value);        return new Entry(key, value);      }      // 進(jìn)行entry的更新      entry.mValue = value;      // 標(biāo)記entry reference引用值為true，意為近期此entry被訪問過，在get,remove方法中，也會更新此屬性值為true      entry.mReferenced = true;      // 標(biāo)記此數(shù)據(jù)為dirty，意為從baking load此entry值后，此值發(fā)生過更新      entry.mDirty = true;      return entry;    });    // 隨后通知Eviction線程，判斷是否需要進(jìn)行entry的移除，在get,remove方法中，也會在末尾調(diào)用此方法    wakeEvictionThreadIfNecessary();  }

文章不錯(cuò)？點(diǎn)個(gè)【在看】吧！??