ElasticSearch核心之——分布式特性

前言

我们知道，ElasticSearch（简称ES）天然支持分布式，具备存储海量数据的能力，其搜索和数据分析的功能都建立在ElasticSearch存储的海量的数据之上。了解ES的分布式的特性可以帮助我们更好的掌握ES的高阶技能，在生产环境中快速地处理ES集群的故障问题。在本篇文章中，我们会对分片、副本、ES集群的故障转移、搭建ES集群、ES实现实时性查询等概念和操作进行讲解，希望能够给各位读者提供一个参考。

一、分布式介绍

（一）分布式的好处

ES支持集群模式，是一个分布式系统，其好处主要有两个∶

1. 增大系统容量，如内存、磁盘，使得es集群可以支持PB级的数据
2. 提高系统可用性，即使部分节点停止服务，整个集群依然可以正常服务.

es集群由多个ES 实例组成。不同集群通过集群名字来区分，可通过cluster.name进行修改，默认为elasticsearch。每个ES实例本质上是一个JVM进程，且有自己的名字，通过node.name进行修改

（二）集群的节点介绍

1. Cluster State 集群状态

ES集群相关的数据称为cluster state，主要记录如下信息∶节点信息，比如节点名称、连接地址等；索引信息，比如索引名称、配置等

2. Master Node 主节点

可以修改cluster state的节点称为master节点，一个集群只能有一个cluster state存储在每个节点上，master维护最新版本并同步给其他节点
master节点是通过集群中所有节点选举产生的，可以被选举的节点称为master-eligible节点，相关配置如下: node.master: true

3. Coordinating Node 处理请求节点

处理请求的节点即为coordinating节点，该节点为所有节点的默认角色，不能取消。路由请求到正确的节点处理，比如创建索引的请求到master节点

4. Data Node 数据存储节点

存储数据的节点即为data节点，默认节点都是data类型，相关配置如下∶node.data: true

二、副本和分片

谈及副本和分片两个概念之前，我们先说一下这两个概念存在的意义：解决系统可用性和增大系统容量
我们想象这样一个场景，我们的数据只存放在一台ES服务器上，那么一旦这台ES出现宕机或者其他不可控因素影响的话，我们除了丧失了服务的可用性外，可能还存在着数据丢失的可能。同时，单机服务的存储容量也无法应对项目对大数据量的要求。

（一）解决系统可用性——副本

系统可用性可以分为服务可用性和数据可用性
服务可用性含义为：当前服务挂掉后，是否有其他服务器顶替当前节点提供服务支持。
数据可用性含义为：当前服务挂掉后，存储在当前服务器上的数据，是否还可以对外提供访问和修改的服务。
副本可以理解为是某个数据的复制体，副本和源数据内容一致。副本的存在可以有效地满足系统可用性的需求，比如说，我们可以在原有节点的基础上复制一个和源节点一模一样的节点，这样一旦原有节点挂掉了，另外一个节点也还是可以替代源节点提供服务，而且复制出来的节点拥有和源节点一样的数据，这样也保障了数据可用性。

（二）增大系统容量——shard分片

我们在上一小节讲到可以使用副本来解决系统可用性的问题，但是这里存在一个问题，不管存在多少个副本（节点），都无法增大源节点的存储空间。在这个问题上，ES引入了Shard分片这个概念来解决问题。

分片是es支持PB级数据的基石

• 分片存储了部分数据，可以分布于任意节点上
• 分片数在索引创建时指定且后续不允许再更改，默认为5个
• 分片有主分片和副本分片之分，以实现数据的高可用
• 副本分片的数据由主分片同步，可以有多个，从而提高读取的吞吐量

看完分片的特点后可能还有人不太清楚到底什么是分片，其实分片是n/1个源节点数据。比如说原ES集群中只有一个主节点，所有的索引数据都存储在这个节点上。现在我们将某个索引数据分成3份，分别存放在3个ES节点上，那么每台ES服务器上就各自有1个分片shard。该索引的所有节点Shard分片的集合，就是索引的全部数据。
下面我们来演示一下：

1. 新建索引，分片为3，副本为1

在cerebro上新建索引index

2. 查看新建的索引分片分布情况

在cerebro上查看新建的索引分片分布情况

为了更好的了解ES的分片机制，大家不妨在上面的案例上进一步思考两个问题：

1. 此时增加节点是否能提高test_shard_index的数据容量?

答案是不能。原因是我们创建索引时定义的分片数量只有3个，且都已经落在了3个节点上。所以即使再增加多一个节点，也不会有对应的Shard分片可以落在新的节点上，并不能扩大test_shard_index的数据容量。

新增节点没有对应的分片

2. 此时增加副本数是否能提高test_shard_index的读取吞吐量?

答案是不能。因为新增的副本也是分布在这3个节点上，还是利用了同样的资源。如果要增加吞吐量，还需要新增节点。

增加副本无法提高节点读取吞吐量

通过上面两个问题，相信大家已经可以认识到分片的重要性，分片数过小，会导致后续无法通过增加节点实现水平扩容；（副本）分片数过大会导致一个节点上分布过多分片，造成资源浪费，同时会影响查询性能

三、集群状态 Cluster Health

集群健康状况，包括以下三种: green健康状态，指所有主副分片都正常分配； yellow指所有主分片都正常分配，但是有副本分片未正常分配； red表示有主分片未分配
我们可以通过这个api查看集群的状态信息： GET _cluster/health

kibana查看集群状态

我们也可以通过cerebro或者head插件来直接获取当前集群的状态

使用cerebro查看集群状态

需要注意的是，即使当前集群的状态为red，也并不代表当前的ES丧失了提供服务的能力。只是说未被分配主分片的索引无法正常存储和操作而已。

四、故障转移

这里故障转移的意思是，当ES集群出现某个或者多个节点宕机的情况，ES实现服务可用性的应对策略。
这里我们新建一个分片为3，副本为1的索引，分片分别分布在三个节点，此时集群为green
当master节点所在机器宕机导致服务终止，此时集群会如何处理呢?

1. node2和node3发现node1无法响应一段时间后会发起master选举，比如这里选择node2为master节点。此时由于主分片PO下线，集群状态变为Red。

node1主节点宕机后集群状态

2. node2发现主分片PO未分配，将RO提升为主分片。此时由于所有主分片都正常分配，集群状态变为Yellow.

node2将副本分片升级为主分片

3. node2 为PO和P1生成新的副本，集群状态变为绿色

node2、node3生成副本分片

我们可以看到，从node1主节点宕机到ES恢复集群可用性的过程中，ES有着自己的故障转移机制，保障了集群的高可用性。我们也可以在自己的本地上去进行试验，建好索引后，kill掉主节点，观察集群状态就行。
同时，此时就算node2宕机了，那么node3也能够很快的恢复服务的提供能力。

五、文档的分布式存储

我们知道，我们创建的文档最终会存储在分片上，那么在分布式集群的基础上，ES集群是怎么判断当前该文档最终应该落在哪一个分片上呢？
很显然，我们需要一个可以实现文档均匀分布到各个分片上的映射算法，那么常见的随机算法和round-robin（轮询）算法可以满足需要吗？答案是不可以，这两个算法虽然可以实现文档均匀分布分片的存储需要，但是当我们通过DocumentId查询文档时，ES并不能知道这个文档ID到底存储在了哪个节点的分片上，所以只能够从所有分片上检索，时间长。如果我们为这个问题建立一个文档和分片映射关系的表，虽然确实可以快速定位到文档对应的存储分片，但是当文档的数据量很大的时候，那么检索的效率也会随之变低。

对于上面这个问题，ES提供的解决方法是建立文档到分片的映射算法
es 通过如下的公式计算文档对应的分片：

shard = hash(routing)% number_of _primary_shards

hash算法保证可以将数据均匀地分散在分片中
routing是一个关键参数，默认是文档id，也可以自行指定
number_of_primary_shards是主分片数
我们可以看到，该算法与主分片数相关，这也是分片数一旦确定后便不能更改的原因

我们已经知道了ES是如何将文档映射到分片上去了，下面我们就来详细讲解一下文档创建、读取的流程。

（一）文档创建的流程

2. node3通过routing 计算该文档应该存储在Shard 1上，查询cluster state后确认主分片P1在node2 上，然后转发创建文档的请求到node2
3. P1接收并执行创建文档请求后，将同样的请求发送到副本分片R1
4. R1接收并执行创建文档请求后，通知P1成功的结果
5. P1接收副本分片结果后，通知node3创建成功
6. node3返回结果到Client

文档创建过程图

（二）文档读取的流程

2. node3通过routing 计算该文档在Shard 1上，查询cluster state后获取 Shard 1的主副分片列表，然后以轮询的机制获取一个shard，比如这里是R1，然后转发读取文档的请求到node1
3. R1接收并执行读取文档请求后，将结果返回node3
4. node3返回结果给Client

文档读取流程图

（三）文档批量创建的流程

2.node3通过 routing计算所有文档对应的shard，然后按照主shard 分配对应执行的操作，同时发
送请求到涉及的主shard，比如这里3个主shard都需要参与
3.主shard接收并执行请求后，将同样的请求同步到对应的副本shard
4.副本shard执行结果后返回结果到主shard，主shard再返回node3
5.node3整合结果后返回Client```
![文档批量创建流程图](https://img.haomeiwen.com/i24009055/14952c1c55953892.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

（四）文档批量读取的流程

2.node3通过routing 计算所有文档对应的shard，然后按照主shard分配对应执行的操作， 分送请求到涉及的主shard，比如这里3个主shard都需要参与
3.主shard接收并执行请求后，将同样的请求同步到对应的副本shard
4.副本shard执行结果后返回结果到主shard，主 shard再返回node3
5.node3整合结果后返回Client

文档批量读取的流程图

六、分布式系统中的脑裂问题

脑裂问题，英文为split-brain，是分布式系统中的经典网络问题，如下图所示:
3个节点组成的集群，突然node1的网络和其他两个节点中断

node1网络中断
node2与node3会重新选举master，比如node2成为了新master，此时会更新cluster state
node1自己组成集群后，也会更新cluster state
同一个集群有两个master，而且维护不同的cluster state，网络恢复后无法选择正确的master
node1重连后，出现了2个master
由于一个集群有2个master的存在，所以这个问题也被叫做脑裂问题。

那么脑裂问题有什么解决办法呢？

解决方案为仅在可选举master-eligible节点数大于等于quorum时才可以进行master选举

1. quorum = master-eligible节点数/2+1，例如3个master-eligible节点时，quorum为2。
2. 设定 discovery.zen.minimum_master_nodes为quorum即可避免脑裂

七、文档搜索实时性

文档搜索的实时性指的是当索引中的文档数据发生变更（新增/修改/删除）后，我们可以实时查询到最新的文档内容。

在讲文档搜索实时性之前，先讲一下倒排索引的不可变更特性。由于倒排索引一旦生成，不可变更的特定，使得其有着以下3点好处：

1. 不用考虑并发写文件的问题，杜绝了锁机制带来的性能问题
2. 由于文件不再更改，可以充分利用文件系统缓存，只需载入一次，只要内存足够，对该文件的读取都会从内存读取，性能高
3. 文件不可变更，使得我们可以对文件格式进行压缩存储，节省磁盘和内存存储空间
   但是这样存在着一个问题，当需要写入新的文档时，我们就必须重新构建倒排索引文件（因为原文件不可以插入新数据），然后替换老文件后，新文档才能被检索，导致文档实时性差。这样的话，如果原索引文件已经非常大的情况下，每次的重构文件都不得不需要消耗大量的时间和资源。
   Lucene的解决方案是新文档直接生成新的倒排索引文件，查询的时候同时查询所有的倒排文件，然后做结果的汇总计算即可
   有新文档则创建一个新的倒排索引文件
   Lucene将构建的单个倒排索引称为segment，合在一起称为Index，与ES中的Index概念不同。ES中的一个Shard对应一个Lucene lndex,Lucene会有一个专门的文件来记录所有的segment信息，称为commit point
   一个segment表示单个倒排索引，合在一起为index

（一）文档搜索实时性——新增文档

下面，将针对Lucene实现文档实时性搜索的几个动作进行讲解，分析其是如何在新增文档后实现ES的搜索实时性的。

1.Reflush

我们从上面的描述中知道，当我们新增了一个文档后会新增一个倒排索引文件segment，但是segment写入磁盘的时间依然比较耗时（难以实现实时性），所以ES借助文件系统缓存的特性，先将segment在缓存中创建并开放查询来进一步提升实时性，该过程在es中被称为refresh。
在refresh之前文档会先存储在一个buffer中，refresh时将 buffer中的所有文档清空并生成 segment
es默认每1秒执行一次refresh，因此文档的实时性被提高到1秒，这也是es被称为近实时(Near Real Time)的原因

ES执行reflush操作

2.Translog

reflush虽然通过将文档存放在缓存中的方式实现了秒级别的实时性，但是如果在内存中的segment还没有写入磁盘前发生了宕机，那么其中的文档就无法恢复了，如何解决这个问题呢?
ES 引入 translog机制。写入文档到buffer时，同时将该操作写入translog中。
translog文件会即时写入磁盘(fsync)，在ES 6.x中，默认每个请求都会落盘，我们也可以修改为每5秒写一次，这样风险便是丢失5秒内的数据，相关配置为index.translog.*。同时ES每次启动时会检查translog 文件，并从中恢复数据。

translog文件的产生

3.Flush

flush 负责将内存中的segment写入磁盘，主要做如下的工作:

1. 将translog写入磁盘
2. 将index buffer清空，其中的文档生成一个新的segment，相当于一个refresh操作
3. 更新commit point并写入磁盘
4. 执行fsync操作，将内存中的segment 写入磁盘
5. 删除旧的translog文件
   Flush的工作过程
   Flush是一个相对来说比较重的操作，我们可以看到一个Flush的操作背后有许多步骤，需要操作到许多资源，所以如果没有特别需要的话，一般不建议频繁使用Flush操作。

Reflush和Flush执行的时机

refresh 发生的时机主要有如下几种情况∶

1. 间隔时间达到时，通过index.settings.refresh_interval来设定，默认是1秒
2. index.buffer占满时，其大小通过indices.memory.index_buffer_size设置，默认为jvm heap的10%，所有shard共享
3. flush 发生时也会发生refresh

flush '发生的时机主要有如下几种情况∶

1. 间隔时间达到时，默认是30分钟，5x之前可以通过index.translog.flush_threshold_period修改，之后无法修改
2. translog 占满时，其大小可以通过index.translog.flush_threshold_size控制，默认是512mb，每个index有自己的translog

（二）文档搜索实时性——删除与更新文档

segment一旦生成后就不能更改，那么如果你要删除文档该如何操作?

1. Lucene专门维护一个.del的文件，记录所有已经删除的文档，注意.del上记录的是文档在Lucene 内部的id
2. 在查询结果返回前会过滤掉.del中的所有文档

更新文档如何进行呢?

ES的做法是首先删除文档，然后再创建新文档

（三）Segment Merging

我们上面提到，新增文档是通过新建segment来解决，删除文档是通过维护.del文件来进行的，假如现在我们设置的reflush时间间隔为1秒，那么一小时单个ES索引就会生成3600个segment，一天下来乃至一个月下来会产生的segment文件数量更是不可想象。为了解决Segment过多可能引起的性能下降问题，ES中采用了Segment Merging（即segment合并）的方法来减少segment的数量。
执行合并操作的方式有两种，一种是ES定时在后台进行Segment Merging操作，还有一种是我们手动执行force_merge_api命令来实现合并操作。

（四）题外补充——ES的Index和Lucene的Index对比

ES的Index和Lucene的Index对比
我们在前面的文字中一直有提到Index这个词，为了不让读者对Index这个概念混淆，所以在此再补充一下Index在ES和Lucene中的含义。我们看上图可以知道，一个或者多个文档组成了一个segment，多个segment（和.del文件）组成了Commit Point，即组成了一个Lucene Index，也就是说Lucene Index相当于ES中的一个分片。而所有的Shard分片则组成了一个ES Index。也即是说，ES Index是多个Lucene index的集合