es分布式架构和原理分析

es的分布式架构

前言

这篇初识ElasticSearch文章中，说明了es的集群的核心概念,回顾一下。
关于节点：
一个运行中的 es实例称为一个节点，而集群是由一个或者多个拥有相同cluster.name 配置的节点组成，它们共同承担数据和负载的压力。当有节点加入集群中或者从集群中移除节点时，集群将会重新平均分布所有的数据。当一个节点被选举成为主节点时，它将负责管理集群范围内的所有变更，例如增加、删除索引，或者增加、删除节点等。而主节点并不需要涉及到文档级别的变更和搜索等操作，所以当集群只拥有一个主节点的情况下，即使流量的增加它也不会成为瓶颈。任何节点都可以成为主节点。作为用户，我们可以将请求发送到集群中的任何节点，包括主节点。每个节点都知道任意文档所处的位置，并且能够将我们的请求直接转发到存储我们所需文档的节点。

分片 shard和副本分片 replica机制：

一个节点包含多个分片。每个分片都是最小工作单元，承载部分数据。
增删节点时，shard会自动在节点中负载均衡。
一个文档只能完整的存放在一个primary shard上,以及对于的relica shard上。
relica shard是primary shard的副本，负责容错，承担读请求。
primary shard的数量在创建索引的时候旧固定了，replica shard的数量可以随时修改。
primary shard不能和自己的replica shard在同一个节点上，因为节点宕机，不能起容错作用。

接下来，学习

容错机制

单节点容错

如果es集群只有一个节点，设置索引(已存在的索引不能设置)的primary shard为3，replica shard为1，那么一个索引有6个shard，但是 primary shard不能和自己的replica shard在同一个节点上，导致replica shard无法分配，这样集群的状态为yellow。

通过 elasticsearch-head 插件查看集群情况。

3 个副本分片都是 Unassigned，它们都没有被分配到任何节点。在同一个节点上既保存原始数据又保存副本是没有意义的，因为一旦失去了那个节点，我们也将丢失该节点上的所有副本数据。当前集群是正常运行的，但存在丢失数据的风险。

结论：单节点的容错性为0。

双节点容错

当集群中只有一个节点在运行时，意味着会有一个单点故障问题。所以我们只需再启动一个节点即可防止数据丢失。当你在同一台机器上启动了第二个节点时，只要它和第一个节点有同样的 cluster.name 配置，它就会自动发现集群并加入到其中。但是在不同机器上启动节点的时候，为了加入到同一集群，你需要配置一个可连接到的单播主机列表。之所以配置为使用单播发现，以防止节点无意中加入集群。只有在同一台机器上运行的节点才会自动组成集群。
设置还是和上面单节点相同。如同双节点的shard情况。

3 个副本分片将会分配到这个节点上——每个主分片对应一个副本分片。这意味着当集群内任何一个节点出现问题时，我们的数据都完好无损。所有新近被索引的文档都将会保存在主分片上，然后被并行的复制到对应的副本分片上。这就保证了我们既可以从主分片又可以从副本分片上获得文档。

结论：容错性为1台。

水平扩容

那同样的设置，再加一个节点，怎么样？如下图：

Node 1 和 Node 2 上各有一个分片被迁移到了新的Node 3节点，现在每个节点上都拥有 2 个分片，而不是之前的 3 个。这表示每个节点的硬件资源（CPU, RAM, I/O）将被更少的分片所共享，每个分片的性能将会得到提升。如果挂了一个节点是怎么样，如node1挂了，数据不会丢失。如果挂了两个节点是怎么样，如node1，node2挂了，数据会1/3丢失，容错性为1台。那不是和两节点的容错一样了？这个时候可以增加副本分片的数量，来提升容错率。在运行中的集群上是可以动态调整副本分片数目的，我们可以按需伸缩集群。让我们把副本数从默认的 1 增加到 2。

但是如果我们想要扩容超过 9 个节点怎么办呢？
主分片的数目在索引创建时就已经确定了下来。实际上，这个数目定义了这个索引能够存储的最大数据量。（实际大小取决于你的数据、硬件和使用场景。）但是，读操作搜索和返回数据可以同时被主分片或副本分片所处理，所以当你拥有越多的副本分片时，也将拥有越高的吞吐量。极限扩容：一个节点上只有一个shard(primary shard和replica shard),增加es读的性能。

Elasticsearch乐观并发控制

乐观并发控制：Elasticsearch 实现乐观锁方案，记录每次更新的版本号，只有拿到最近的版本号的更新操作，才可以更新成功。其他拿到过期的版本号，一般重试。

ES新版本不使用version进行并发版本控制 if_seq_no=版本值&if_primary_term=文档位置

插入一条数据

PUT /es_db/_doc/7
{"name":"caicai","sex":0,"age":22,"address":"长沙","remark":"pythonassistant"
}

查询文档信息

GET /es_db/_doc/7//结果
{"_index" : "es_db","_type" : "_doc","_id" : "7","_version" : 3,"_seq_no" : 10,"_primary_term" : 1,"found" : true,"_source" : {"name" : "caicai","sex" : 0,"age" : 22,"address" : "长沙","remark" : "pythonassistant"}
}

更新

POST /es_db/_update/7?if_seq_no=10&&if_primary_term=1
{"doc": {"name":"cc"}
}

doc的增删改流程

es为了实现实时搜索，在写入

当索引一篇文档时发什么？

选择任意一个数据节点发送请求，这个数据节点就成为一个coordinating node（协调节点）。
根据文档id的散列值选择一个分片，并将文档发送到该分片。这份分片可能位于另一个节点。

shard = hash(routing) % number_of_primary_shards

coordinating node会进行路由，将请求转发给对应的primary shard所在的DataNode。
然后文档被发送到该主分片的所有副文本进行索引，使得主副分片之间保持数据同步。数据同步可以让副分片提供搜索服务，并在原主分片不能提供服务时，升级为主分片。

routing是一个可变值，默认是文档的 _id ，也可以设置成一个自定义的值。 routing 通过hash 函数生成一个数字，然后这个数字再除以 number_of_primary_shards （主分片的数量）后得到余数。这个分布在 0 到 number_of_primary_shards-1 之间的余数，就是我们所寻求的文档所在分片的位置。

如下图
这就解释了为什么我们要在创建索引的时候就确定好主分片的数量并且永远不会改变这个数量:因为如果数量变化了，那么所有之前路由的值都会无效，文档也再也找不到了。

可选的请求参数允许您影响这个过程，可能以数据安全为代价提升性能。这些选项很少使用，因为 Elasticsearch 已经很快，但是为了完整起见，请参考下文：

consistency:即一致性。在默认设置下，即使仅仅是在试图执行一个写操作之前，主分片都会要求必须要有规定数量quorum的分片副本处于活跃可用状态，才会去执行写操作。这是为了避免在发生网络分区故障的时候进行写操作，进而导致数据不一致。规定数量即： int((primary + number_of_replicas) / 2 ) + 1
consistency 参数的值可以设为：

one ：只要主分片状态 ok 就允许执行写操作。
all：必须要主分片和所有副本分片的状态没问题才允许执行写操作。
quorum：默认值为quorum , 即大多数的分片副本状态没问题就允许执行写操作。
注意，规定数量的计算公式中number_of_replicas指的是在索引设置中的设定副本分片数，而不是指当前处理活动状态的副本分片数。如果你的索引设置中指定了当前索引拥有3个副本分片，那规定数量的计算结果即：int((1 primary + 3 replicas) / 2) + 1 = 3，如果此时你只启动两个节点，那么处于活跃状态的分片副本数量就达不到规定数量，也因此您将无法索引和删除任何文档。

timeout:如果没有足够的副本分片会发生什么？Elasticsearch 会等待，希望更多的分片出现。默认情况下，它最多等待 1 分钟。如果你需要，你可以使用timeout参数使它更早终止：100是100 毫秒，30s是30秒。

搜索分片时发送了什么？

某个DataNode接收到请求，该DataNode就会成为协调节点（Coordinating Node）;
协调节点（Coordinating Node）将查询请求广播到每一个数据节点，这些数据节点的分片会处理该查询请求;
每个分片进行数据查询，将符合条件的数据放在一个优先队列中，并将这些数据的文档ID、节点信息、分片信息返回给协调节点；
协调节点将所有的结果进行汇总，并进行全局排序；
协调节点向包含这些文档ID的分片发送get请求，对应的分片将文档数据返回给协调节点，最后协调节点将数据返回给客户端。

注意：已经被索引的文档可能已经存在于主分片上但是还没有复制到副本分片。在这种情况下，副本分片可能会报告文档不存在，但是主分片可能成功返回文档。一旦索引请求成功返回给用户，文档在主分片和副本分片都是可用的。

更新分片时发送了什么？

客户端向DataNode发送更新请求，该DataNode就会成为协调节点（Coordinating Node）;。
请求转发到主分片所在的节点。
数据节点主分片检索文档，修改_source字段中的JSON，并且尝试重新索引主分片的文档。如果文档已经被另一个进程修改,它会重试 ,超过retry_on_conflict次后放弃。
如果成功地更新文档，它将新版本的文档并行转发到其他的副本分片，重新建立索引。一旦所有副本分片都返回成功，该数据节点向协调节点也返回成功，协调节点向客户端返回成功。

Elasticsearch准实时索引实现

溢写到文件系统缓存

es为了实现实时搜索，在写入doc时利用了buffer(内存)，OS cache（系统缓存）,disk（磁盘）。es底层时lucene实现的，而在luncene中一个索引会分成多个数据段（sement）,每一个数据端都会存放索引的部分doc。es把一个索引中的doc分散存在到多个shard(主分片)，在每个shared中又使用多个segment来存储数据。

客户端发起请求（增删改）到es中，es将本次请求要操作的doc写入到buffer中，然后通过内存的buffer生成一个segment，并刷到文件系统缓存中。es为了保证搜索的近实时(NRT)，默认每秒刷新一次buffer，这个刷新时间间隔可以手动修改，也可以用命令刷新，建议刷新数据设置在1秒左右，好处服务器宕机后，只丢失一秒左右的数据。如果buffer里面没有任何数据，则不会执行刷新机制。

写translog保障容错

es将doc写入buffer的同时，也会将内容写入到translog文件中，这份文件存在的意义在于即使es宕机，也能尽可能减少数据丢失。translog也不能保证数据绝对不丢失。
translog在系统重启后，es会重新读取磁盘中保持的数据（sement）到缓存中，接着读取translog中的操作日志并逐行执行。
translog文件的持久有两种方式，默认时per request fsync每次客户端写入请求被持久化以后，才会回应200（后台也会异步每5s持久化一次）。

translog flush原理
translog会不断的增大，在内存中积压数量众多的index segment file的文件流也在不断增大，os cache中积压的数据也越来越大。当translog 文件大到一定程度或者30分钟执行一次，es会自动触发commit操作。

将内存中数据刷新到index segment中，index segment写入os cache并打开index segemt为搜索服务。
执行一个commit point操作，将os cache中所有的segment记录在commit point中，并持久化到磁盘disk。
commit point操作会触发fsync操作，将内存中已经写入数据的index segment强制刷新到disk，持久化成文件。
清空本次持久化的index segment对应在translog中的日志。

segment合并

每一秒会生成一个index segment文件，每30分钟就会index segment文件持久化到磁盘，Segment太多时，ES定期会将多个segment合并成为大的segment，减少索引查询时IO开销，此阶段ES会真正的物理删除（之前执行过的delete的数据）。

es会选取大小相近的segment文件流，合并成一个大的segment。
执行commit操作，在磁盘中记录commit point，这个commit不仅包含新增的segment，还包含需要删除的segment。
commit操作结束后，es会将合并的segment文件重新打开，为搜索提供服务，需要被删除的segment文件则进行关闭。

详情流程

提升集群读取性能的方法

利用缓存机制，减少不必要的算分。
避免使用*开头的通配符查询
优化分片
- 避免Over Sharing:一个查询需要访问每一个分片，分片过多，会导致不必要的查询开销
- 控制单个分片的大小Search: 20GB Logging: 40GB

提升写入性能的方法

写性能优化的目标:增大写吞吐量，越高越好
- 客户端:多线程，批量写可以通过性能测试，确定最佳文档数量
- 服务器端:降低IO操作,降低CPU和存储开销,尽可能做到写入和分片的均衡负载，实现水平扩展