Solr In Action 筆記(3) 之 SolrCloud基礎

Solr In Action 筆記(3) 之 SolrCloud基礎

在Solr中，一個索引的實例稱之為Core，而在SolrCloud中，一個索引的實例稱之為Shard；Shard 又分為leader和replica。

1. SolrCloud的特質

作為分布式搜索引擎的SolrCloud具有以下幾個特質：

• 可擴展性

      所謂的可擴展性就是指可以通過擴大集群的規模來實現性能的提升。有兩種方式來實現可擴展性，一種是縱向擴展，即加快CPU速度，增加RAM，提升磁盤I/O性能等，另一種是橫向擴展，就是分布式系統中使用的通過增加節點來擴大集群規模。SolrCloud采用的時橫向擴展，可以通過兩個方式來實現擴展，一種是增加shard，它實現的是將大數據量的index切分成多個小的index，這種方式是從將每個shard的數據量的角度來提升SolrCloud的性能的，另一種是增加replica，即增加shard的備份的數量，這種方式的優點是增加SolrCloud的並發查詢能力以及容災能力，至於原因在接下來會具體闡述。基本上對於SolrCloud來說，它的計算能力隨着資源(即shard)的增加成正比。

     以下是Solrcloud擴展性具有的限制以及改善方法。

Scalability / limitation	Mitigation strategy
Number of documents indexed: Having a large number of documents in an index impacts the performance of faceting, sorting, and constructing filters. Also, we are currently limited to 2.1 billion documents per Lucene index due to the document ID being stored as an integer.	Split large indexes into multiple smaller indexes using sharding;
Document size and complexity: Having many fields or large text fields requires more memory and faster disk I/O.	Add more RAM and faster disks.
Indexing throughput: You may need to index thousands of documents per second.	Distribute indexing operations across multiple nodes using sharding.
Document volatility: If existing documents change frequently, your indexes will be more volatile, requiring constant seg- ment merging.	Get faster disks to facilitate constant seg- ment merging;
Query volume (typically measured by QPS—queries per second).	Use replication to increase the number of threads available to execute queries.
Query complexity: This includes facets, grouping, custom sorting impact, and query execution performance.	Use sharding and replication to parallel- ize complex query computations such as faceting and sorting.

 

• 高可用性

    高可用性主要還是指SolrCloud的容災能力，SolrCloud使用leader和replica以及交叉備份的方式實現數據的冗余以實現很好的容災性能，leader和replica是同一shard的相同數據，存放在不同的主機上。當某台主機宕機時，該主機的數據在另外的主機上具有備份，這樣就確保了數據的不丟失，同時SolrCloud在進行查詢的時候是只對認准shard而不區分leader和replica，所以即使有一台主機宕機了也不會影響SolrCloud的查詢結果，它只是稍微影響了查詢性能。

• 一致性

     對於一個分布式搜索引擎來說，一致性，性能，以及分割容差是三個主要指標，其中一致性與讀寫性能是個矛盾的指標，以SolrCloud為例，SolrCloud選擇了一致性而適當放棄了寫的性能。SolrCloud具有replica時，當有數據建立索引，SolrCloud首先將數據update至leader shard，然后leadershard再將數據進行分發至各個replica shard，leader shard進行分發是個同步的過程，也就是說它會一直等到所以replica shard的數據update成功才會返回成功，中間一旦出現錯誤就視為失敗，這樣就充分保證了leader和replica的數據一致性，當然這也就降低了寫的速度。這里需要說明的是，當replica是不上線狀態時候，SolrCloud的leader是不會分發至這個replica shard的，關於shard 的狀態在下文中將會具體介紹。至於為什么SolrCloud對弱一致性的零容忍態度，主要是避免索引的部分成功以及多個shard查詢結果的不同。

• 簡單性

當有主機recovering失敗了，當我們處理完失敗的原因后將該主機上線，SolrCloud會自動從leader那進行數據同步。

• 伸縮性

SolrCloud支持將一份shard分成兩份小的shard

2. Zookeeper

    SolrCloud使用zookeeper主要實現以下三點功能：

• 集中配置存儲以及管理
• 集群狀態改變時進行監控以及通知
• shard leader選舉

    關於zookeeper的具體介紹請看《Zookeeper之基礎學習》。

2.1 zookeeper 數據類型

    Zookeeper的組織結構是類似於文件系統，如下圖所示，每一個層是一個Znode，每一個Znode存儲了一些元數據例如創建時間，修改時間以及一些小量的數據。這里需要指出，Zookeeper並不支持存放大數據數據，它只支持小於1M大小的數據，這是因為性能原因，Zookeeper將數據存放在內存中。

      Zookeeper另一個重要的概念是短鏈接，當Zookeeper客戶端與Zookeeper建立一個短連接后會在Zookeeper新建一個Znode，客戶端會一直與Zookeeper進行通信並保證這個Znode一直存在。如果當客戶端與Zookeeper的短連接斷開，這個Znode就會消失。在SolrCloud中，/live_nodes下就會存在所有客戶端的短連接，表示現在有哪些Solr組成SolrCloud。

     當Solr跟Zookeeper保持短連接時，這些Solr主機就組成了SolrCloud，如果其中一個Solr的短連接斷掉了，那么Live_nodes下就少了一個Znode，SolrCloud也就少了一個主機，於是Zookeeper就會告訴其他剩余的Solr有一個Solr掛掉了，等會進行查詢以及leader數據分發時候不用再經過剛才那個Solr了。那么Zookeeper是如何知道有Solr掛了呢，這就是下面要講的watcher

2.2 Znode Watcher

     Znode Watcher是Zookeeper一個重要概念和特性，當Znode發生變化時候，Zookeeper會告訴所有在該Znode注冊的Watcher這個Znode改變了，從而出發相應的事務。比如SolrCloud的/clusterstate.json  Znode,SolrCloud的所有的Solr都會在這里進行注冊，如果當有新的replica或者node下線了，所有的watcher就會收到通知，然后每一個Solr就會更新它的clusterstate。這在SolrCloud的shard leader選舉過程中將會詳細介紹。

2.3 配置

     Zookeeper同樣是個分布式系統，同樣具有容災能力和擴展能力。首先簡要描述下集群的容災的能力，Zookeeper集群能有效的運行的基本要求是它的實際運行節點的數量大於集群總數量的一半，也就是說如果實際部署的Zookeeper的集群節點個數為5個，那么當出現故障時候，正常的Zookeeper節點個數至少3個才能保證Zookeeper繼續正常運行。所有一般情況下，Zookeeper集群的節點個數都為奇數(2N+1,N為至少存活的節點個數)。

     Zookeeper被建議獨立部署，不要使用SolrCloud的內置zookeeper以及和Solr一起使用。當Zookeeper下線時候，經測試發現，雖然SolrCloud無法正常啟動，建索引功能異常，但是由於每一個Solr節點保存了之前zookeeper的狀態以及配置信息，我們仍然可以對SolrCloud進行查詢，且結果和性能正常。

     SolrCloud配置Zookeeper時候，需要在solr.xml內的zkHost加入zookeeper節點配置，如zk1.example.com:2181, zk2.example.com:2181 。

2.4 Zookeeper Client Timeout

     當Solr加入集群后，會在Znode上建立短連接，當Solr下線時候，Zookeeper就會刪除這個Znode。我們希望Zookeeper能夠盡可能快的察覺出SolrCloud的集群狀態變化。默認情況下，這個時間是15秒，我們可以設置zkClientTimeout 這個配置來修改他。

     請注意，當JVM進行full垃圾回收的時候會暫停所有正在運行的線程，當然也包括zookeeper的線程。如果full垃圾回收所持續的時間大於zkClientTimeout，那么就會出現SolrCloud以為zookeeper已經下線而無法正常工作，所以如果zookeeper出現異常下線，請去查看垃圾回收的日志來排除是否這個引起的。

2.5  Zookeeper的配置存儲以及分布式管理

     Zookeeper在SolrCloud的一個重要的功能是存儲SolrCloud的配置以及對配置進行分布式管理。我們可以將配置信息(比如solrconfig.xml和schmea.xml)上傳至Zookeeper，如果這些配置信息發生變化Zookeeper就會將這些新的配置文件更新到集群中每一個Solr節點上，如果有新的Solr加入集群，那么Zookeeper就會傳遞給他一份配置文件，Zookeeper很好的完成了一個配置文件的存儲以及分布式管理的功能。Zookeeper上存儲的文件最大為1M，關於Zookeeper的具體介紹請看另外的文章。

3 Shard 和 Replica

3.1 Shard和Replica的數量

      增加Shard和Replica的數量能提升SolrCloud的查詢性能和容災能力，但是我們仍然得根據實際的document的數量，document的大小，以及建索引的並發，查詢復雜度，以及索引的增長率來統籌考慮Shard和Replica的數量。

3.2 集群管理

      正常的SolrCloud Index 和Query都需要集群的shard 和 replica 都處於active狀態， shard leader需要知道它的所有的active Replica才能進行update操作。SolrCloud具有多種狀態：

狀態	描述
Active	Active nodes are happily serving queries and accepting update requests. Active rep- licas are in sync with their shard leader. A healthy cluster is one in which all nodes are active.
Inactive	Used during shard splitting to indicate that a Solr instance is no longer participating in the collection. Shards that get split enter this state once the splits are active.
Construction	Used during shard splitting to indicate that a split is being created. Shards in this state buffer update requests from the parent shard but do not participate in queries.
Recovering	Recovering instances are running but can’t serve queries. They do accept update requests while recovering so that they don’t continue to fall behind the leader.
Recovery Failed	The instance attempted to recover but encountered an error. In most cases, you will need to consult the logs and manually resolve the issue preventing the instance from recovering.
Down	The instance is running and is connected to ZooKeeper but is not in a state in which it can recover, such as when Solr is initializing. A downed instance does not partici- pate in queries or accept updates. The down state is usually temporary, and the node will transition to one of the other states.
Gone	The instance is not connected to ZooKeeper and has probably crashed. If a node is still running but ZooKeeper thinks it’s gone, the most likely cause is an OutOfMemoryError in the JVM.

      Solr依賴Zookeeper實現集群的管理，在Zookeeper中有一個Znode 是/clusterstate.json  ，它存儲了當前時刻下整個集群的狀態。同時在一個集群中有且只會存在一個overseer，如果當前的overseer fail了那么SolrCloud就會選出新的一個overseer，就跟下文要講的shard leader 選取類似。

      每一個Index 實例，無論是leader還是replica都會在/clusterstate.json上注冊一個watcher以便於接受集群狀態變更的通知，當有一個新的Solr節點加入Zookeeper時候，overseer就會更新修改/clusterstate.json，一旦/clusterstate.json發生修改，Zookeeper就會通知所有的Index 實例，讓他們更新集群信息的緩存。同理其他的集群狀態變更。

3.3 Shard leader 選舉

      Shard leader的選舉對於SolrCloud來說還是很重要的過程，雖然Shard leader選舉並不直接影響SolrCloud的查詢，但是卻大大影響了SolrCloud的建索引過程。關於SolrCloud的Index過程將在下一節介紹。Shard leader在index過程中主要起到以下作用：

• 接受shard的update請求
• 在需要的update的document中加入_version_域，並實施 optimistic lock
• 將document寫入update log
• 並行發送document到所有的replica，直到replica返回完成相應才結束

    首先，任何node 包括replica都可以被自動選舉為leader，leader的選舉同樣依靠Zookeeper。選舉過程如下：

啟動時候：

• 當Index實例(shard)上線時，它會嘗試短連接至zookeeper，並在Zookeeper創建對應shard的Znode，比如/collections/logmill/leader_select/shard1/election/XXX_node1_0000001,其中logmill表示collection的名字，shard1表示是shard的編號，XXX_node1_0000001表示最終創建的Znode的名字，XXX我們暫時不關心。
• 如果我們設置多個Replica時候，那么一個shard就會有多個Index實例去短連接至zookeeper，並同樣在Zookeeper上創建shard的Znode，比如/collections/logmill/leader_select/shard1/election/XXX_node1_0000002。
• Zookeeper的短連接成功后，會給Znode創建一個編號，這個編號是同步自增的，也就是說，多個短連接請求時候Zookeeper會保證處理完一個后再處理另一個，所以不同的短連接生成的Znode編號是遞增的且不會重疊的。由於受到節點自身的性能以及網絡等影響，不同節點短連接存在先后順序從而造成Znode編號不同，比如XXX_node1_0000001，XXX_node1_0000002，那么SolrCloud選舉Shard Leader的策略很簡單，即始終保持leader_select/shard1/目錄下編號最小的為leader，其他的為replica，比如XXX_node1_0000001為leader，XXX_node1_0000002為replica。
• 選好leader后，會在/collections/logmill/leader/shard1下生成leader的信息。
• Replica知道自己不是leader后，它會在leader的節點上(比如XXX_node1_0000001)注冊一個watcher，該watcher一直監控leader的狀態。

 上述過程是正常啟動SolrCloud的leader選舉過程，那么接下來來了解下leader下線觸發leader重新選舉：

leader重新選舉：

• 正常運行的SolrCloud已產生一個leader(Znode編號最小，比如XXX_node1_0000001)，后續的Replica后在leader節點上注冊Watcher。當Leader下線時候，即短連接斷開，那么Zookeeper上的Znode(比如XXX_node1_0000001)就會被刪除。
• 此時，所有Replica在Leader節點上的watcher就會監控到這一變化，所有的Replica就會進行leader選舉，選舉的原則依然是判斷自己是不是目前注冊在/collections/logmill/leader_select/shard1/election下的Znode編號最小的那位，是的話就是Leader，否則就是Replica
• 如果判斷自己是Replica，就會繼續在leader的Znode上(這個時候的leader是XXX_node1_0000002)注冊watcher，等待leader下線再次觸發選舉leader。
• 如果這個時候原先下線的leader上線了會怎么樣，它就會被當做新的一個Solr節點注冊到Zookeeper上，並獲取一個比現有Znode更大的編號，在Leader Znode節點上注冊watcher，等待它的選舉機會。

leader的選舉還是蠻簡單的，下圖是leader選舉好的效果圖：

leader選舉源碼

    leader選舉過程主要在LeaderElector類中。

• joinElection 主要實現shard節點在Zookeeper上建立短連接，並生產Znode，獲取Znode編號。

public int joinElection(ElectionContext context, boolean replacement) throws KeeperException, InterruptedException, IOException {
    context.joinedElectionFired();
    //select Znode節點的路徑
    final String shardsElectZkPath = context.electionPath + LeaderElector.ELECTION_NODE;
    
    long sessionId = zkClient.getSolrZooKeeper().getSessionId();
    String id = sessionId + "-" + context.id;
    String leaderSeqPath = null;
    boolean cont = true;
    int tries = 0;
    while (cont) {
      try {
        //在Zookeeper創建短連接，並創建Znode，生成Znode編號
        leaderSeqPath = zkClient.create(shardsElectZkPath + "/" + id + "-n_", null,
            CreateMode.EPHEMERAL_SEQUENTIAL, false);
        //Znode路徑
        context.leaderSeqPath = leaderSeqPath;
        cont = false;
      //如果短連接建立失敗，進行多次嘗試。
      } catch (ConnectionLossException e) {
        // we don't know if we made our node or not...
        List<String> entries = zkClient.getChildren(shardsElectZkPath, null, true);
        
        boolean foundId = false;
        for (String entry : entries) {
          String nodeId = getNodeId(entry);
          if (id.equals(nodeId)) {
            // we did create our node...
            foundId  = true;
            break;
          }
        }
        if (!foundId) {
          cont = true;
          if (tries++ > 20) {
            throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
                "", e);
          }
          try {
            Thread.sleep(50);
          } catch (InterruptedException e2) {
            Thread.currentThread().interrupt();
          }
        }

      } catch (KeeperException.NoNodeException e) {
        // we must have failed in creating the election node - someone else must
        // be working on it, lets try again
        if (tries++ > 20) {
          throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
              "", e);
        }
        cont = true;
        try {
          Thread.sleep(50);
        } catch (InterruptedException e2) {
          Thread.currentThread().interrupt();
        }
      }
    }
    //Znode的編號
    int seq = getSeq(leaderSeqPath);
    //開始進行leader選舉
    checkIfIamLeader(seq, context, replacement);
    
    return seq;
  }

• checkIfIamLeader 開始真正的leader選舉，根據Zookeeper上創建的Znode的編號大小判斷自己是否是leader，如果是replica，則在leader的Znode上注冊watcher，等到再次進行leader選舉

private void checkIfIamLeader(final int seq, final ElectionContext context, boolean replacement) throws KeeperException,
      InterruptedException, IOException {
    context.checkIfIamLeaderFired();
    // get all other numbers...
    final String holdElectionPath = context.electionPath + ELECTION_NODE;
    //獲取現有的select Znode節點下已注冊的所有的Znode
    List<String> seqs = zkClient.getChildren(holdElectionPath, null, true);
    //對所有shard的Znode按Znode編號進行從小到大排序
    sortSeqs(seqs);
    List<Integer> intSeqs = getSeqs(seqs);
    if (intSeqs.size() == 0) {
      log.warn("Our node is no longer in line to be leader");
      return;
    }
    //如果自己的shard的Znode編號是最小的，那么就進行自己就是leader
    if (seq <= intSeqs.get(0)) {
      // first we delete the node advertising the old leader in case the ephem is still there
      try {
        zkClient.delete(context.leaderPath, -1, true);
      } catch(Exception e) {
        // fine
      }

      runIamLeaderProcess(context, replacement);
    } else {
      //如果自己的shard的Znode編號不是最小的，那么自己就是replica，則在Znode找出誰是leader
      // I am not the leader - watch the node below me
      int i = 1;
      for (; i < intSeqs.size(); i++) {
        int s = intSeqs.get(i);
        if (seq < s) {
          // we found who we come before - watch the guy in front
          break;
        }
      }
      int index = i - 2;
      if (index < 0) {
        log.warn("Our node is no longer in line to be leader");
        return;
      }
     //找出leader后，在leader的Znode上注冊watcher，監視leader狀態。
      try {
        zkClient.getData(holdElectionPath + "/" + seqs.get(index), watcher = new ElectionWatcher(context.leaderSeqPath , seq, context) , null, true);
      } catch (KeeperException.SessionExpiredException e) {
        throw e;
      } catch (KeeperException e) {
        log.warn("Failed setting watch", e);
        // we couldn't set our watch - the node before us may already be down?
        // we need to check if we are the leader again
        checkIfIamLeader(seq, context, true);
      }
    }
  }

 

最后將下SolrCloud的leader策略，一開始一直蠻鄙視SolrCloud的這個leader策略，也曾花了一段時間來重寫leader策略，該策略的效果是使得每一個shard的leader均勻分布在每一台服務器上，也就是說每一台服務器只會有一個leader，這樣的效果就是無論查詢還是建索引，每一台服務器的負載是完全均衡的，而且如果leader掛機再上線后，該leader能馬上重新變成leader而不會成為Znode編號最大的Replia。隨着對SolrCloud的理解的深入，越來越覺得這個leader策略的妙處以及我的修改的錯誤，好處在前文已經詳細介紹了。所以建議對於這個leader策略沒必要還是不要去修改了。