對於一個分散式儲存系統來說，資料是分散儲存在多個節點上的。如何讓資料均衡的分佈在不同節點上，來保證其高可用性？所謂均衡，是指系統中每個節點的負載是均勻的，並且在發現有不均勻的情況或者有節點增加/刪除時，能及時進行調整，保持均勻狀態。本文將探討Elasticsearch的資料分佈方法，文中所述的背景是Elasticsearch 5.5。
  在Elasticsearch中，以Shard為最小的資料分配/遷移單位。資料到節點的對映分離為兩層：一層是資料到Shard的對映（Route），另一層是Shard到節點的對映（Allocate）。

  一方面，插入一條資料時，ES會根據指定的Key來計算應該落到哪個Shard上。預設Key是自動分配的id，可以自定義，比如在我們的業務中採用CompanyID作為Key。因為Primary Shard的個數是不允許改變的，所以同一個Key每次算出來的Shard是一樣的，從而保證了準確定位。

shard_num = hash(_routing) % num_primary_shards

  另一方面，Master會為每個Shard分配相應的Data節點進行儲存，並維護相關元資訊。通過Route計算出來的Shard序號，在元資訊中找到對應的儲存節點，便可完成資料分佈。Shard Allocate的對映關係並不是完全不變的，當檢測到資料分佈不均勻、有新節點加入或者有節點掛掉等情況時就會進行調整，稱為Relocate。那麼，Elasticsearch是根據什麼規則來為Shard選取節點，從而保證資料均衡分佈的？概括來看，主要有三方面的影響：節點位置、磁碟空間、單個節點上的Index和Shard個數

節點位置

   對於一個ES節點來說，它可能是某臺物理機器上的一個VM，而這個物理機器位於某個Zone的某個機架（Rack)上。通過將Primary Shard和Replica Shard分散在不同的物理機器、Rack、Zone，可以儘可能的降低資料丟失和系統不可用的風險，這一點幾乎在所有的分散式系統中都會考量。

   Elasticsearch是通過設定awareness.attribute對叢集中的節點進行分組，從而實現Rack和Zone的發現。比如按照下列方式對elasticsearch.yml進行配置，再啟動相應的節點，即可實現Zone的區分。

// elasticsearch.yml
cluster.routing.allocation.awareness.attributes: zone

// 啟動ES
./bin/elasticsearch -Enode.attr.zone=zone_one
./bin/elasticsearch -Enode.attr.zone=zone_two
 

  實踐中，如果使用了這樣的Awareness機制，應該保證不同分組類的機器個數一致，不會發生傾斜。比如，在Zone Awareness下，如果叢集有10臺機器，應該保證每個Zone各有5臺機器（2個Zone）。

磁碟空間

  磁碟空間是制約儲存的硬性條件，單機的可用磁碟空間決定了能否繼續往這個節點寫入新資料、分配新Shard以及是否需要遷移資料等。在ES中，有三個引數用來控制與此相關的特性，預設每30秒檢查一次。

• cluster.routing.allocation.disk.watermark.low: 預設為85%，超過這個閾值後，就不允許往這個節點分配Shard。
• cluster.routing.allocation.disk.watermark.high：預設為90%，超過這個閾值後，就需要將該節點的Shard遷移出去。
• cluster.routing.allocation.disk.watermark.flood_stage：預設為95%，超過這個閾值後，與該節點上的Shard有關的Index都變成只讀，不允許寫入資料。

單個節點上的Index和Shard個數

  在滿足節點位置和磁碟空間的條件後，單個節點上的Index和Shard個數是否均勻，決定了Shard可以分配/遷移到哪個節點。ES通過計算權值來量化這樣的分配方式。
  以檢測某個Shard是否需要遷移到其他節點為例，ES會先計算該Shard所在節點（A）的權值，然後依次跟其他節點的權值比較，如果與節點B的差值（Delta-A）超過了閾值，再進一步計算節點A去掉該Shard後的權值與節點B增加該Shard後的權值之間的差值（Delta-B），如果Delta-A大於Delta-B，則表明Shard可以遷移到節點B。
  這裡的權值計算簡化如下，其中indexBalance與shardBalance分別由引數控制，而閾值由cluster.routing.allocation.balance.threshold設定，預設為1.0f。當然，這裡只描述了核心思想，詳細邏輯請閱讀BalancedShardsAllocator.java中的原始碼。通過調整三個引數，可以控制策略的鬆緊。

// indexBalance = cluster.routing.allocation.balance.index, default is 0.55f
// shardBalance = cluster.routing.allocation.balance.shard, default is 0.45f

float sum = indexBalance + shardBalance;
float theta0 = shardBalance / sum;
float theta1 = indexBalance / sum;

private float weight(Balancer balancer, ModelNode node, String index, int numAdditionalShards) {
    final float weightShard = node.numShards() + numAdditionalShards - balancer.avgShardsPerNode();
    final float weightIndex = node.numShards(index) + numAdditionalShards - balancer.avgShardsPerNode(index);
    return theta0 * weightShard + theta1 * weightIndex;
}

Primary與Replica分佈

  最初關注Elasticsearch的資料分佈，是因為在效能調優時遇到了一個與Primary/Replica分佈有關的問題。背景是這樣的，為了能夠複用單個節點上的Disk Cache，我們對查詢請求進行了限制，只允許其訪問Primary Shard。然而總是有那麼一兩臺機器的查詢會被Queue住，通過調研發現，這些機器上面的Primary Shard比其他機器多（對某一個Index而言），即下圖中左邊所示，而我們希望的是右圖所示的均勻分佈。
  引起這個問題的根源是，Elasticsearch中的Shard均勻分佈是針對Primary+Replica整體而言的，也就是說沒法做到只針對Primary Shard單方面做均勻分佈，所以才會出現下圖左邊所示，某個節點上有3個Primary Shard，而另一個節點只有1個。目前尚未發現可以調節的地方。

  本文探討了Elasticsearch的資料分佈方法，其思想對很多其他分散式儲存系統是通用的，而瞭解相關原理是做很多調優工作的前提。

（全文完，本文地址：https://blog.csdn.net/zwgdft/article/details/83478241 ）
（版權宣告：本人拒絕不規範轉載，所有轉載需徵得本人同意，並且不得更改文字與圖片內容。大家相互尊重，謝謝！）

Bruce
2018/10/30 晚