Elasticsearch原理学习--分片原理及分片路由转发

浅浅的花香味﹌ 2022-05-20 01:48 240阅读 0赞

### 一、分片与副本设置 ###

**1、分片（shard）**

Elasticsearch集群允许系统存储的数据量超过单机容量，实现这一目标引入分片策略shard。在一个索引index中，数据（document）被分片处理（sharding）到多个分片上。Elasticsearch屏蔽了管理分片的复杂性，使得多个分片呈现出一个大索引的样子。

**2、副本（replica）**

为了提升访问压力过大是单机无法处理所有请求的问题，Elasticsearch集群引入了副本策略replica。副本策略对index中的每个分片创建冗余的副本，处理查询时可以把这些副本当做主分片来对待（primary shard），此外副本策略提供了高可用和数据安全的保障，当分片所在的机器宕机，Elasticsearch可以使用其副本进行恢复，从而避免数据丢失。

**3、设置**

`number_of_shards`

每个索引的主分片数，默认值是 `5` 。这个配置在索引创建后不能修改。

`number_of_replicas`

每个主分片的副本数，默认值是 `1` 。对于活动的索引库，这个配置可以随时修改。

例如，我们可以创建只有 一个主分片，没有副本的小索引（只对某个索引有效）：

PUT /my_temp_index
    {
        "settings": {
            "number_of_shards" :   1,
            "number_of_replicas" : 0
        }
    }

然后，我们可以用 `update-index-settings` API 动态修改副本数：

PUT /my_temp_index/_settings
    {
        "number_of_replicas": 1
    }

默认分片数为5、副本为1：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70][] 分片与副本

### 4、查看索引分片与副本信息 ###

（1）查看所有索引分片与副本信息：http://localhost:9200/\_settings

返回所有索引的信息

{
        "index2": {
            "settings": {
                "index": {
                    "creation_date": "1547693736615",
                    "number_of_shards": "5",
                    "number_of_replicas": "1",
                    "uuid": "-jpUoiQKQYCIL_vc9RWTpA",
                    "version": {
                        "created": "6010399"
                    },
                    "provided_name": "index2"
                }
            }
        },
        "index": {
            "settings": {
                "index": {
                    "creation_date": "1547462626951",
                    "number_of_shards": "5",
                    "number_of_replicas": "1",
                    "uuid": "psUlEdttRLGjzxXe81A0wg",
                    "version": {
                        "created": "6010399"
                    },
                    "provided_name": "index"
                }
            }
        }
    }

（2）查看某个索引的分片与副本信息:http://localhost:9200/\{索引名称\}/\_settings

http://localhost:9200/index/\_settings 结果如下：

{
        "index": {
            "settings": {
                "index": {
                    "creation_date": "1547462626951",
                    "number_of_shards": "5",
                    "number_of_replicas": "1",
                    "uuid": "psUlEdttRLGjzxXe81A0wg",
                    "version": {
                        "created": "6010399"
                    },
                    "provided_name": "index"
                }
            }
        }
    }

### 二、分片算法 ###

Elasticsearch默认的分片个数为5，对于同一个索引的不同id的数据持久化需要进行统一的分片处理，简单来说分片算法就是计算id的hash值，然后与分片数进行floorMod运算(类似取余)。

OperationRouting的generateShardId方法：

public static int generateShardId(IndexMetaData indexMetaData, @Nullable String id, @Nullable String routing) {
            final String effectiveRouting;
            final int partitionOffset;
    
            if (routing == null) {
                assert(indexMetaData.isRoutingPartitionedIndex() == false) : "A routing value is required for gets from a partitioned index";
                effectiveRouting = id;
            } else {
                effectiveRouting = routing;
            }
    
            if (indexMetaData.isRoutingPartitionedIndex()) {
                partitionOffset = Math.floorMod(Murmur3HashFunction.hash(id), indexMetaData.getRoutingPartitionSize());
            } else {
                // we would have still got 0 above but this check just saves us an unnecessary hash calculation
                partitionOffset = 0;
            }
    
            return calculateScaledShardId(indexMetaData, effectiveRouting, partitionOffset);
        }
    
        private static int calculateScaledShardId(IndexMetaData indexMetaData, String effectiveRouting, int partitionOffset) {
            //计算hash值
            final int hash = Murmur3HashFunction.hash(effectiveRouting) + partitionOffset;
    
            // we don't use IMD#getNumberOfShards since the index might have been shrunk such that we need to use the size
            // of original index to hash documents
            //类似取余运行获取分片结果
            return Math.floorMod(hash, indexMetaData.getRoutingNumShards()) / indexMetaData.getRoutingFactor();
        }

### 三、节点分片信息 ###

集群中三个节点的分片：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 1][] 标题

节点分片信息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 2][] 标题

请求转发：

TransportReplicationAction中会根据索引的分片结果找到分片所在的节点，如果在本地则直接进行索引操作，如果在其他节点则将请求转发到其他节点中。

ReroutePhase的doRun方法：

@Override
            protected void doRun() {
                setPhase(task, "routing");
                final ClusterState state = observer.setAndGetObservedState();
                if (handleBlockExceptions(state)) {
                    return;
                }
    
                // request does not have a shardId yet, we need to pass the concrete index to resolve shardId
                final String concreteIndex = concreteIndex(state);
                final IndexMetaData indexMetaData = state.metaData().index(concreteIndex);
                if (indexMetaData == null) {
                    retry(new IndexNotFoundException(concreteIndex));
                    return;
                }
                if (indexMetaData.getState() == IndexMetaData.State.CLOSE) {
                    throw new IndexClosedException(indexMetaData.getIndex());
                }
    
                // resolve all derived request fields, so we can route and apply it
                resolveRequest(indexMetaData, request);
                assert request.shardId() != null : "request shardId must be set in resolveRequest";
                assert request.waitForActiveShards() != ActiveShardCount.DEFAULT : "request waitForActiveShards must be set in resolveRequest";
    
    			//找到节点路由表
                final ShardRouting primary = primary(state);
                if (retryIfUnavailable(state, primary)) {
                    return;
                }
    			//根据分片id找到节点信息
                final DiscoveryNode node = state.nodes().get(primary.currentNodeId());
    			//本地节点则本地执行
                if (primary.currentNodeId().equals(state.nodes().getLocalNodeId())) {
                    performLocalAction(state, primary, node, indexMetaData);
                } else {
    				//远程节点则发送请求到远程
                    performRemoteAction(state, primary, node);
                }
            }

参考文章

[elasticsearch shard split 思考][elasticsearch shard split]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70]: /images/20220520/93134dc918714300a8e805ec97e19e6a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 1]: /images/20220520/93d0247585b94f7a8dcb7ce0a62f941f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70 2]: /images/20220520/33e497b3d554441d8c2d77764da7ec8b.png
[elasticsearch shard split]: https://www.jianshu.com/p/cfdfefbe30a2