從這篇文章開始終于要討論比較正常版本的hbase了---0.92.1~~

Scan是hbase提供的非常重要的功能之一，我們的hbase分析系列就從這兒開始吧。

首先，我們有一些background的知識需要了解:

1.HBase是如何區分不同記錄的，大家可以參考http://punishzhou.iteye.com/blog/1266341，講的比較詳細

2.Region，MemStore，Store，StoreFile分別的含義和如何工作的，可以參考淘寶的入門文檔http://www.searchtb.com/2011/01/understanding-hbase.html

3.Scan的客戶端實現可以參考http://punishzhou.iteye.com/blog/1297015

4.hbase客戶端如何使用scan，這個文章實在太多了，隨便搜一篇吧~

在這篇文章中，我們主要focus在Scan的Server端實現。

Scan的概念是掃描數據集中[startkey，stopkey)的數據，數據必須是全體有序的，根據hbase mem+storefile的結構我們大致描述下scan的步驟：

1.準備好所有的scanner，包括memstorescanner，storefilescanner

2.將scanner放入一個prorityQueue

3.開始scan，從prorityQueue中取出當前所有scanner中最小的一個數據記錄

4.如果3取出的滿足結果則返回，如果不滿足則從prorityQueue中取next

5.如果取出的數據記錄等于stopkey或者prorityQueue為空則結束

下面開始進入實現部分。

scan過程中會使用到諸多scanner，scanner類圖如下：

步驟1：準備Scanner：

memstore和storefile的getScanner：

return Collections.<KeyValueScanner>singletonList(
          new MemStoreScanner());

List<StoreFileScanner> scanners = new ArrayList<StoreFileScanner>(
        files.size());
    for (StoreFile file : files) {
      StoreFile.Reader r = file.createReader();
      scanners.add(r.getStoreFileScanner(cacheBlocks, usePread, isCompaction));
    }
    return scanners;

store層的getScanner：對memstore和storefile分別判斷是否滿足scan的條件，包括時間，是否支持bloomfilter

List<KeyValueScanner> allStoreScanners =
        this.store.getScanners(cacheBlocks, isGet, false);

    List<KeyValueScanner> scanners =
        new ArrayList<KeyValueScanner>(allStoreScanners.size());

    // include only those scan files which pass all filters
    for (KeyValueScanner kvs : allStoreScanners) {
      if (kvs instanceof StoreFileScanner) {
        if (memOnly == false
            && ((StoreFileScanner) kvs).shouldSeek(scan, columns)) {
          scanners.add(kvs);
        }
      } else {
        // kvs is a MemStoreScanner
        if (filesOnly == false && this.store.memstore.shouldSeek(scan)) {
          scanners.add(kvs);
        }
      }
    }

Store層：獲取store的scanner們，seekExactly檢查是否符合我們scan的column，scanner.seek調用memscanner或者storefilescanner的seek，檢查我們查詢的startkey是否在當前的scanner范圍中，過濾掉不需要搜索的查詢，其中，storefilescanner會使用bloomfilter來seek。當收集到當前store的scanner們后會構建store層的KeyValueHeap。

// pass columns = try to filter out unnecessary ScanFiles
    List<KeyValueScanner> scanners = getScanners(scan, columns);

    // Seek all scanners to the start of the Row (or if the exact matching row
    // key does not exist, then to the start of the next matching Row).
    if (matcher.isExactColumnQuery()) {
      for (KeyValueScanner scanner : scanners)
        scanner.seekExactly(matcher.getStartKey(), false);
    } else {
      for (KeyValueScanner scanner : scanners)
        scanner.seek(matcher.getStartKey());
    }

    // Combine all seeked scanners with a heap
    heap = new KeyValueHeap(scanners, store.comparator);

region層：將store中的scanner中取出來放到scanners里，并創建RegionScanner的KeyValueHeap

for (Map.Entry<byte[], NavigableSet<byte[]>> entry :
          scan.getFamilyMap().entrySet()) {
        Store store = stores.get(entry.getKey());
        StoreScanner scanner = store.getScanner(scan, entry.getValue());
        scanners.add(scanner);
      }
      this.storeHeap = new KeyValueHeap(scanners, comparator);

是不是已經被諸多scanner看暈了，這邊先梳理下思路：

• memscanner和storefilescanner為直接數據交互的scanner，因此繼承KeyValueScanner接口，每次next讀取一個KeyValue對象
• storescanner管理里面的memscanner和storefilescanner，并且將所有的scanner放到一個KeyValueHeap內，KeyValueHeap會保證每次next都會取Store中滿足條件的最小值
• regionscanner類似storescanner，管理所有的storescanner，并將所有的scanner放到KeyValueHeap中，作用同上

2.初始化KeyValueHeap（其實在第一部中已經做了），關鍵是初始化內部的PriorityQueue

this.comparator = new KVScannerComparator(comparator);
    if (!scanners.isEmpty()) {
      this.heap = new PriorityQueue<KeyValueScanner>(scanners.size(),
          this.comparator);
      for (KeyValueScanner scanner : scanners) {
        if (scanner.peek() != null) {
          this.heap.add(scanner);
        } else {
          scanner.close();
        }
      }
      this.current = heap.poll();

關于PriorityQueue可以看看http://blog.csdn.net/hudashi/article/details/6942789，內部實現了一個heap。

3.開始scan的next方法，首先，peekRow取得KeyValueHeap中當前的rowkey。這是通過current的scanner peek獲得當前的rowkey，從第二部可知，KeyValueHeap剛開始時current即為heap中最小的那個

public KeyValue peek() {
    if (this.current == null) {
      return null;
    }
    return this.current.peek();
  }

4.開始取出符合當前rowkey的values
調用heap.next 循環從heap中取出相同key的不同value，直到heap取出的key不等于當前的key為止，這就表示我們已經遍歷到下一個rowkey了必須停止這次next操作。

do {
            this.storeHeap.next(results, limit - results.size());
            if (limit > 0 && results.size() == limit) {
              if (this.filter != null && filter.hasFilterRow()) {
                throw new IncompatibleFilterException(
                  "Filter with filterRow(List<KeyValue>) incompatible with scan with limit!");
              }
              return true; // we are expecting more yes, but also limited to how many we can return.
            }
          } while (Bytes.equals(currentRow, nextRow = peekRow()));

heap的next方法：首先取出當前的scanner，調用next方法取出一個result塞到results里，然后peek判斷這個scanner是否沒數據了，如果沒了就關閉，如果有就再將scanner放入heap中，再取出下一個最小的scanner

if (this.current == null) {
      return false;
    }
    InternalScanner currentAsInternal = (InternalScanner)this.current;
    boolean mayContainMoreRows = currentAsInternal.next(result, limit);
    KeyValue pee = this.current.peek();
    /*
     * By definition, any InternalScanner must return false only when it has no
     * further rows to be fetched. So, we can close a scanner if it returns
     * false. All existing implementations seem to be fine with this. It is much
     * more efficient to close scanners which are not needed than keep them in
     * the heap. This is also required for certain optimizations.
     */
    if (pee == null || !mayContainMoreRows) {
      this.current.close();
    } else {
      this.heap.add(this.current);
    }
    this.current = this.heap.poll();
    return (this.current != null);

5.對取出的results進行filter，判斷是否到了stoprow，如果到了next返回false

	final boolean stopRow = isStopRow(nextRow);

          // now that we have an entire row, lets process with a filters:

          // first filter with the filterRow(List)
          if (filter != null && filter.hasFilterRow()) {
            filter.filterRow(results);
          }

          if (results.isEmpty() || filterRow()) {
            // this seems like a redundant step - we already consumed the row
            // there're no left overs.
            // the reasons for calling this method are:
            // 1. reset the filters.
            // 2. provide a hook to fast forward the row (used by subclasses)
            nextRow(currentRow);

            // This row was totally filtered out, if this is NOT the last row,
            // we should continue on.

            if (!stopRow) continue;
          }
          return !stopRow;

最后總結一下:

1.scan的實現是非常復雜的，原因主要是因為hbase在內存和硬盤中有很多顆有序樹，scan時需要將多顆有序樹merge成一個

2.scan.next出來的list<KeyValue>是同一個key下按照一定順序從小到大排列的，順序是key>column>quality>timestamp>type>maxsequenceId,然后如果是memstore，則比較memstoreTs，大的排前面，而且memstore的maxsequenceId默認是整數最大值

3.最好能指明scan的cf和quality，這樣會加快速度

4.memstore的scan和storefile的scan如果有機會后面會再寫文詳細闡述