Elasticsearch 模塊 - Shard Allocation 機制

時間  2021-03-18
標籤 html java node web json 服務器 網絡 併發 app 負載均衡 欄目 日誌分析 简体版
原文   原文鏈接

原文html

1. 背景

shard allocation 意思是分片分配, 是一個將分片分配到節點的過程; 可能發生該操做的過程包括:java

  • 初始恢復(initial recovery)
  • 副本分配(replica allocation)
  • 從新平衡(rebalance)
  • 節點的新增和刪除

來源node

分片的分配操做, 是由 master 角色的節點來決定何時移動分片, 以及移動到哪一個節點上, 以達到集羣的均衡;web

說明json

本文基於 Elasticsearch 7.4.0 版本服務器

2. 機制分析

2.1. Allocation 觸發條件

  • 新增或刪除 index 索引
  • node 節點的新增或刪除
  • 執行 reroute 命令
  • 修改 replica 副本數量
  • 集羣重啓

具體對應源碼解釋:來源網絡

序號 調用函數 說明
1 AllocationService.applyStartedShards Shard 啓動狀態修改
2 AllocationService.applyFailedShards Shard 失效狀態修改
3 AllocationService.deassociateDeadNodes Node 節點離開集羣
4 AllocationService.reroute(AllocationCommands) 執行 relocation 命令
5 TransportClusterUpdateSettingsAction.masterOperation 集羣配置修改操做
6 MetaDataCreateIndexService.onlyCreateIndex 建立新索引 index 請求
7 MetaDataDeleteIndexService.deleteIndexs 刪除索引 index 操做
8 MetaDataIndexStateService.closeIndex 關閉 index 操做
9 MetaDataIndexStateService.openIndex 打開 index操做
10 NodeJoinController.JoinTaskExecutor 經過集羣發現的節點加入集羣
11 GatewayService.GatewayRecoveryListener 經過 GatewayRecovery 恢復的節點加入集羣
12 LocalAllocateDangledIndices.submitStateUpdateTask 恢復磁盤內存而在 MateDate 內不存在的 index
13 RestoreService.restoreSnapshot 從 snapshot 中恢復的 index

2.2. Rebalance 的觸發條件

在 rebalance 以前會通過 2.3.2 中介紹的全部策略裏實現的 canRebalance 方法, 所有經過後纔會執行下面的 Rebalance 過程;併發

Rebalance 過程是經過調用 balanceByWeights() 方法, 計算 shard 所在的每一個 node 的 weight 值,app

\[weightShard = node.numShards() + numAdditionalShards - balancer.avgShardsPerNode() \\ weightIndex = node.numShards(index) + numAdditionalShards - balancer.avgShardsPerNode(index) \\ weight = theta0 * weightShard + theta1 * weightIndex \\ \]

其中:負載均衡

  • numAdditionalShards 通常爲 0, 調用 weightShardAdded, weightShardRemoved 方法時分別取值爲 1 和 -1;
  • theta0 = cluster.routing.allocation.balance.shard 系統動態配置項, 默認值爲 0.45f;
  • theta1 = cluster.routing.allocation.balance.index 系統動態配置項, 默認值爲 0.55f;

權重計算公式

源碼以下:

private static class WeightFunction {
    private final float indexBalance;
    private final float shardBalance;
    private final float theta0;
    private final float theta1;
    WeightFunction(float indexBalance, float shardBalance) {
        float sum = indexBalance + shardBalance;
        if (sum <= 0.0f) {
            throw new IllegalArgumentException("Balance factors must sum to a value > 0 but was: " + sum);
        }
        theta0 = shardBalance / sum;
        theta1 = indexBalance / sum;
        this.indexBalance = indexBalance;
        this.shardBalance = shardBalance;
    }
    float weight(Balancer balancer, ModelNode node, String index) {
        final float weightShard = node.numShards() - balancer.avgShardsPerNode();
        final float weightIndex = node.numShards(index) - balancer.avgShardsPerNode(index);
        return theta0 * weightShard + theta1 * weightIndex;
    }
}

2.3. 源碼分析

分片分配就是把一個分片分配到集羣中某個節點的過程, 其中分配決策包含了兩個方面:

  • 哪些分片應該分配到哪些節點上
  • 哪一個分片做爲主分片, 哪一個做爲副本分片

Elasticsearch 主要經過兩個基礎組件來完成分片分配這個過程的: allocator 和 deciders;

  • allocator 尋找最優的節點來分配分片;
  • deciders 負責判斷並決定是否要進行分配;
  1. 新建的索引

​ allocator 負責找出擁有分片數量最少的節點列表, 按分片數量遞增排序, 分片數量較少的會被優先選擇; 對於新建索引, allocator 的目標是以更爲均衡的方式把新索引的分片分配到集羣的節點中;

​ deciders 依次遍歷 allocator 給出的節點列表, 判斷是否要把分片分配給該節點, 好比是否知足分配過濾規則, 分片是否將超出節點磁盤容量閾值等等;

  1. 已有的索引

​ allocator 對於主分片, 只容許把主分片指定在已經擁有該分片完整數據的節點上; 對於副本分片, 則是先判斷其餘節點上是否已有該分片的數據的拷貝, 若是有這樣的節點, allocator 則優先把分片分配到這其中一個節點上;

2.3.1. Allocator

Allocator

  • PrimaryShardAllocator 找到擁有某 Shard 最新數據(主分片)的節點;
  • ReplicaShardAllocator 找到磁盤上擁有這個 Shard 數據(副本分片)的節點;
  • BalancedShardsAllocator 找到擁有最少 Shard 個數的節點;
public class BalancedShardsAllocator implements ShardsAllocator {
    public static final Setting<Float> INDEX_BALANCE_FACTOR_SETTING = Setting.floatSetting("cluster.routing.allocation.balance.index", 0.55f, 0.0f, Property.Dynamic, Property.NodeScope);
    public static final Setting<Float> SHARD_BALANCE_FACTOR_SETTING = Setting.floatSetting("cluster.routing.allocation.balance.shard", 0.45f, 0.0f, Property.Dynamic, Property.NodeScope);
    public static final Setting<Float> THRESHOLD_SETTING = Setting.floatSetting("cluster.routing.allocation.balance.threshold", 1.0f, 0.0f, Property.Dynamic, Property.NodeScope);

    private volatile WeightFunction weightFunction;
    private volatile float threshold;
}

2.3.2. Deciders

Deciders 決策期基礎組件的抽象類爲 AllocationDecider:

public abstract class AllocationDecider {
    public Decision canRebalance(ShardRouting shardRouting, RoutingAllocation allocation) {
        return Decision.ALWAYS;
    }
    public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
        return Decision.ALWAYS;
    }
    public Decision canRemain(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
        return Decision.ALWAYS;
    }
    public Decision canAllocate(ShardRouting shardRouting, RoutingAllocation allocation) {
        return Decision.ALWAYS;
    }
    public Decision canAllocate(IndexMetadata indexMetadata, RoutingNode node, RoutingAllocation allocation) {
        return Decision.ALWAYS;
    }
    public Decision canAllocate(RoutingNode node, RoutingAllocation allocation) {
        return Decision.ALWAYS;
    }
    public Decision shouldAutoExpandToNode(IndexMetadata indexMetadata, DiscoveryNode node, RoutingAllocation allocation) {
        return Decision.ALWAYS;
    }
    public Decision canRebalance(RoutingAllocation allocation) {
        return Decision.ALWAYS;
    }
}

ES 7.4.0 中的 Decider 決策器包括如下所示, 他們均實現上面的 AllocationDecider 抽象類, 並重寫 canRebalance, canAllocate, canRemain, canForceAllocatePrimary 等方法;

AllocationDecider

決策器比較多, 大體分類以下, 並列舉決策器對應的配置項:

2.3.2.1. 負載均衡類

  • SameShardAllocationDecider: 避免主副分片分配到同一個節點;

  • AwarenessAllocationDecider: 感知分配器, 感知服務器, 機架等, 儘可能分散存儲 Shard;

    對應的配置參數有:

    cluster.routing.allocation.awareness.attributes: rack_id

    cluster.routing.allocation.awareness.attributes: zone

  • ShardsLimitAllocationDecider: 同一個節點上容許存在同一個 index 的 shard 數目;

    index.routing.allocation.total_shards_per_node: 表示該索引每一個節點上容許最多的 shard 數量; 默認值=-1, 表示無限制;

    cluster.routing.allocation.total_shards_per_node: cluster 級別, 表示集羣範圍內每一個節點上容許最多的 shard 數量, 默認值=-1, 表示無限制;

    index 級別會覆蓋 cluster 級別;

2.3.2.2. 併發控制類

  • ThrottlingAllocationDecider: recovery 階段的限速配置, 避免過多的 recovering allocation 致使該節點的負載太高;

    cluster.routing.allocation.node_initial_primaries_recoveries: 當前節點在進行主分片恢復時的數量, 默認值=4;

    cluster.routing.allocation.node_concurrent_incoming_recoveries: 默認值=2, 一般是其餘節點上的副本 shard 恢復到該節點上;

    cluster.routing.allocation.node_concurrent_outgoing_recoveries: 默認值=2, 一般是當前節點上的主分片 shard 恢復副本分片到其餘節點上;

    cluster.routing.allocation.node_concurrent_recoveries: 統一配置上面兩個配置項;

  • ConcurrentRebalanceAllocationDecider: rebalace 併發控制, 表示集羣同時容許進行 rebalance 操做的併發數量;

    cluster.routing.allocation.cluster_concurrent_rebalance, 默認值=2

    經過檢查 RoutingNodes 類中維護的 reloadingShard 計數器, 看是否超過配置的併發數;

  • DiskThresholdDecider: 根據節點的磁盤剩餘量來決定是否分配到該節點上;

    cluster.routing.allocation.disk.threshold_enabled, 默認值=true;

    cluster.routing.allocation.disk.watermark.low: 默認值=85%, 達到這個值後, 新索引的分片不會分配到該節點上;

    cluster.routing.allocation.disk.watermark.high: 默認值=90%, 達到這個值後, 會觸發已分配到該節點上的 Shard 會 rebalance 到其餘節點上去;

2.3.2.3. 條件限制類

  • RebalanceOnlyWhenActiveAllocationDecider: 全部 Shard 都處於 active 狀態下才能夠執行 rebalance 操做;

  • FilterAllocationDecider: 經過接口動態設置的過濾器; cluster 級別會覆蓋 index 級別;

    index.routing.allocation.require.{attribute}
    index.routing.allocation.include.{attribute}
    index.routing.allocation.exclude.{attribute}
    cluster.routing.allocation.require.{attribute}
    cluster.routing.allocation.include.{attribute}
    cluster.routing.allocation.exclude.{attribute}

    • require 表示必須知足, include 表示能夠分配到指定節點, exclude 表示不容許分配到指定節點;
    • {attribute} 還有 ES 內置的幾個選擇, _name, _ip, _host;

  • ReplicaAfterPrimaryActiveAllocationDecider: 保證只在主分片分配完成後(active 狀態)纔開始分配副本分片;

  • ClusterRebalanceAllocationDecider: 經過集羣中 active 的 shard 狀態來決定是否能夠執行 rebalance;

    cluster.routing.allocation.allow_rebalance

    indices_all_active(默認): 當集羣全部的節點分配完成, 才能夠執行 rebalance 操做;
    indices_primaries_active: 只要全部主分片分配完成, 才能夠執行 rebalance 操做;
    always: 任何狀況下都容許 rebalance 操做;

  • MaxRetryAllocationDecider: 防止 shard 在失敗次數達到上限後繼續分配;

    index.allocation.max_retries: 設置分配的最大失敗重試次數, 默認值=5;

2.3.2.4. 其餘決策類

  • EnableAllocationDecider: 設置容許分配的分片類型; index 級別配置會覆蓋 cluster 級別配置;

    all(默認): 容許全部類型的分片;

    primaries: 僅容許主分片;

    new_primaries: 僅容許新建索引的主分片;

    none: 禁止分片分配操做;

  • NodeVersionAllocationDecider: 檢查分片所在 Node 的版本是否高於目標 Node 的 ES 版本;

  • SnapshotInProgressAllocationDecider: 決定 snapshot 期間是否容許 allocation, 由於 snapshot 只會發生在主分片上, 因此該配置只會限制主分片的 allocation;

    cluster.routing.allocation.snapshot.relocation_enabled

接下來介紹一下在 Elasticsearch 中涉及到 Allocation 和 Rebalance 的相關配置項;

3. cluster-level 配置

3.1. Shard allocation 配置

控制分片的分配和恢復;

配置 默認值 說明
cluster.routing.allocation.enable all 啓用或禁用針對特定類型分片的分配;
1. all: 容許分配全部類型的分片;
2. primaries: 只容許分配主分片(primary shard);
3. new_primaries: 只容許分配新索引的主分片(primary shard);
4. none: 禁用分片分配;
該設置不會影響重啓節點時本地主分片的恢復;
cluster.routing.allocation.node_concurrent_incoming_recoveries 2 一個節點容許併發的傳入分片(incoming shard)數量
cluster.routing.allocation.node_concurrent_outgoing_recoveries 2 一個節點容許併發的傳出分片(incoming shard)數量
cluster.routing.allocation.node_concurrent_recoveries 上面二者的合併配置
cluster.routing.allocation.node_initial_primaries_recoveries 4 單個節點上同時初始化的主分片數量
cluster.routing.allocation.same_shard.host false 是否執行檢查, 以防止基於host name和host address, 在單個主機上分配同一分片的多個實例; 該設置僅用於在同一臺計算機上啓動多個節點的狀況;

3.2. Shard rebalancing 配置

控制集羣之間的分片平衡;

配置 默認值 說明
cluster.routing.rebalance.enable all 啓用或禁用針對特定類型分片的rebalancing;
1. all: 容許rebalancing全部類型的分片;
2. primaries: 只容許rebalancing主分片;
3. replicas: 只容許rebalancing副本分片;
4. none: 禁用rebalancing;
cluster.routing.allocation.allow_rebalance indices_all_active 指定什麼時候容許執行rebalancing;
1. always: 老是容許;
2. indices_primaries_active: 當集羣中全部主分片已分配時才容許rebalancing;
3. indices_all_active: 當集羣中全部分片(包括主分片和副本分片)都已分配時才容許rebalancing;
cluster.routing.allocation.cluster_concurrent_rebalance 2 指定整個集羣中容許同時在節點間移動的分片數量; 該配置僅控制因爲集羣不平衡引發的併發分片分配數量, 對分配過濾(allocation filtering)或強制感知(forced awareness)的分片分配不作限制;

3.3. 分片平衡啓發式

如下配置用於決定每一個分片的存放位置; 當rebalancing操做再也不使任何節點的權重超過balance.threshold時, 集羣即達到平衡;

配置 默認值 說明
cluster.routing.allocation.balance.shard 0.45f 定義節點上分配的分片總數的權重因子; 提高該值會致使集羣中全部節點趨向於分片數量相等;
cluster.routing.allocation.balance.index 0.55f 定義節點上分配的每一個索引的分片數量的權重因子; 提高該值會致使集羣中全部節點上每一個索引的分片數量趨向於相等;
cluster.routing.allocation.balance.threshold 1.0f 定義應當執行操做的最小優化值(非負浮點數); 提高該值會致使集羣在優化分片平衡方面不太積極;

4. Index-level 配置

如下配置控制每一個索引中的分片分配;

4.1. index-level 分片分配過濾(來源)

配置須要分兩步:

  1. 在每一個 Elasticsearch 節點的 elasticsearch.yml 配置文件中添加自定義節點屬性, 好比以 small, medium, big區分節點類型, 則配置文件中可添加:
node.attr.size: medium

或者在啓動 Elasticsearch 服務時, 在命令行裏添加 ./bin/elasticsearch -Enode.attr.size=medium;

  1. 在新建索引的 mapping 時, 添加index.routing.allocation.include/exclude/require.size: medium 的過濾配置便可;
PUT <index_name>/_settings
{
  "index.routing.allocation.include.size": "medium"
}

能夠配置多個自定義節點屬性, 而且必須同時知足索引裏配置的多個過濾條件;

  • index.routing.allocation.include.{attribute}: {values}
  • index.routing.allocation.require.{attribute}: {values}
  • index.routing.allocation.exclude.{attribute}: {values}

其中 {attribute} 能夠是上面提到的自定義節點屬性, ES 本身也有一些內置的節點屬性:

attribute 說明
_name 經過節點名稱進行匹配
_host_ip 經過節點 IP 地址進行匹配
_publish_ip 經過節點的發佈 IP 地址進行匹配
_ip 經過 _host_ip 或 _publish_ip 進行匹配
_host 經過節點的hostname進行匹配
_id 經過節點的 id 進行匹配

其中 {values} 能夠是單個值, 也能夠是逗號分隔的多個值, 也可使用通配符 * 進行模糊匹配;

4.2. 設置延遲分配, 當節點離開時(來源)

當某個節點因爲突發緣由, 好比網絡中斷, 人爲操做重啓等, 須要暫時離開集羣時, 集羣會馬上新建副本分片以替換丟失的副本, 而後在剩餘的全部節點之間進行rebalancing, 這樣致使在短期內該突發節點又恢復過來後, 原先的副本就沒法再使用, 集羣會將剛纔新建的副本分片再拷貝回到該節點上; 這樣就會形成沒必要要的資源浪費, 以及節點分片rebalancing帶來的波動;

可使用 index.unassigned.node_left.delayed_timeout 動態設置來延遲因爲節點離開而致使未分配的副本分片的分配問題; 該配置默認值 1m;

PUT _all/_settings
{
  "settings": {
    "index.unassigned.node_left.delayed_timeout": "5m"
  }
}

修改爲以上配置後, 若是在 5m 內, 該節點能夠恢復從新加入集羣, 則集羣會自動恢復該節點的副本分片分配, 恢復速度很快;

注意

  1. 此設置不影響將副本分片升級爲主分片;
  2. 此設置不影響以前未分配的副本分片;
  3. 在整個集羣從新啓動後, 該延遲分配不會生效;

4.3. 索引恢復的優先級(來源)

索引分片恢復的優先級按照:

  • 可選的 index.priority 配置, 值越大優先級越高;
  • index 索引的建立日期, 越新的索引優先級越高;
  • index 索引的名稱;

4.4. 每一個節點的分片總數(來源)

配置 默認值 說明
index.routing.allocation.total_shards_per_node unbounded(-1) 指定單個節點上最多分配的分片數量, 包括主分片和副本分片;(具體某個索引)
cluster.routing.allocation.total_shards_per_node unbounded(-1) 指定單個節點上最多分配的分片數量, 包括主分片和副本分片;(與索引無關, 全局設置)

這些配置是硬性配置, 可能會致使一些分片沒法分配, 須要慎重配置;

5. 閱讀來源

  1. Shard allocation and cluster-level routing
  2. Elasticsearch底層系列之 Shard Allocation 機制
  3. ELASTICSEARCH ALLOCATION 分析
  4. Elasticsearch Shard Allocation 機制
  5. Elasticsearch Allocation&Rebalance
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程