一、前言：

今天来学习下 es 的写入原理。

Elasticsearch底层使用Lucene来实现doc的读写操作：

Luence 存在的问题：

没有并发设计
lucene只是一个搜索引擎库，并没有涉及到分布式相关的设计，因此要想使用Lucene来处理海量数据，并利用分布式的能力，就必须在其之上进行分布式的相关设计。
非实时
将文件写入lucence后并不能立即被检索，需要等待lucene生成一个完整的segment才能被检索
数据存储不可靠
写入lucene的数据不会立即被持久化到磁盘，如果服务器宕机，那存储在内存中的数据将会丢失
不支持部分更新
lucene中提供仅支持对文档的全量更新，对部分更新不支持。例如：对文档进行部分更新，只新增一个字段或者修改某一字段的值，Lucene是不支持的。

二、Elasticsearch的写入方案

针对Lucene的问题，ES做了如下设计

2.1 shard：

为了支持对海量数据的存储和查询，需要用到分布式系统，通过大规模集群来提高系统水平扩展能力，因此Elasticsearch引入分片的概念，一个索引被分成多个分片（shard）。

除了将index 分片以提高水平扩展能力，Elasticsearch还会将shard复制成多个副本，放置到不同的机器上，提高系统可用性，并且副分片还提供读服务，分担集群压力。

每个shard都是一个lucene段，是可以独立执行搜索任务最小单位。

但是多副本也会带来一致性问题。部分副本写成功，部分副本写失败。

在这里插入图片描述

例如：下面的集群由三个节点组成。存在一个索引，有两个主分片，每个主分片有两个副本分片。相同分片的副本不会放在同一节点。

在这里插入图片描述

Elasticsearch采用多Shard方式，通过路由规则将数据分成多个数据子集，每个数据子集提供独立的索引和搜索功能。当写入文档的时候，根据routing规则，将文档发送给特定Shard中建立索引。这样就能实现分布式了。

如何确定一条数据属于哪个shard？

ES会根据公式：

 shard_num = hash(_routing) % num_primary_shards

_routing的默认值是文档的_id

通过计算得出文档要分配到的分片，在从集群元数据中找出对应主分片的位置，将请求路由到该分片进行文档写操作。

2.2 近实时性-refresh操作

当一个文档写入Lucene后是不能被立即查询到的，Elasticsearch提供了一个refresh操作，为内存中新写入的数据生成一个新的segment，此时被处理的文档均可以被检索到。refresh操作的时间间隔由refresh_interval参数控制，默认为1s, 当然还可以在写入请求中带上refresh表示写入后立即refresh，另外还可以调用refresh API显式refresh。

2.3 部分更新

lucene支持对文档的整体更新，ES为了支持局部更新，在Lucene的Store索引中存储了一个_source字段，该字段的key值是文档ID，内容是文档的原文。当进行更新操作时先从_source中获取原文，与更新部分合并后，再调用lucene API进行全量更新，对于写入了ES但是还没有refresh的文档，可以从translog中获取。另外为了防止读取文档过程后执行更新前有其他线程修改了文档，ES增加了版本机制，当执行更新操作时发现当前文档的版本与预期不符，则会重新获取文档再更新。

三、写入操作：

分别从集群角度和 shard 自身角度来介绍数据如何写入。

3.1 集群角度: Primary -> Replica

我们可以发送请求到集群中的任一节点。每个节点都有能力处理任意请求。每个节点都知道集群中任一文档位置，所以可以直接将请求转发到需要的节点上。

在这里插入图片描述

客户端向NODE1 发送写请求。
检查Active的Shard数。
NODE1 使用文档ID来确定文档属于的分片（图例是：分片0），通过集群状态中的信息获知分片0的主分片位于NODE3，因此请求被转发到NODE3上。
NODE3上的主分片执行写操作。
并发的向所有同步副本发起写入请求，将请求并行转发到NODE1和NODE2的副分片上。
等待所有同步副本返回结果，返回成功或者失败后，返回给Client。

（1）为什么要检查Active的Shard数？

ES中有一个参数，叫做wait_for_active_shards。这个参数的含义是，在每次写入前，该shard至少具有的active副本数。假设我们有一个Index，其每个Shard有3个Replica，加上Primary则总共有4个副本。如果配置wait_for_active_shards为3，那么允许最多有一个Replica挂掉，如果有两个Replica挂掉，则Active的副本数不足3，此时不允许写入。
这个参数默认是1，即只要Primary在就可以写入。如果配置大于1，可以起到一种保护的作用，保证写入的数据具有更高的可靠性。但是这个参数只在写入前检查，并不保证数据一定在至少这些个副本上写入成功，所以并不是严格保证了最少写入了多少个副本。

（2）写入Primary完成后，为何要等待所有同步Replica响应(或连接失败)后返回？

早期ES版本，Primary和Replica之间是允许异步复制的，即写入Primary成功即可返回。但是这种模式下，如果Primary挂掉，就有丢数据的风险，而且从Replica读数据也很难保证能读到最新的数据。所以后来ES就取消异步模式了，改成Primary等同步Replica返回后再返回给客户端。

https://github.com/elastic/elasticsearch/blob/master/docs/reference/docs/data-replication.asciidoc
Once all in-sync replicas have successfully performed the operation and responded to the primary, the primary acknowledges the successful completion of the request to the client.
{
    "_shards" : {
        "total" : 2,
        "failed" : 0,
        "successful" : 2
    }
}

（3）如果某个Replica持续写失败，用户是否会经常查到旧数据？

假如一个Replica持续写入失败，那么这个Replica上的数据可能落后Primary很多。Primary会将这个信息报告给Master，然后Master会在Meta中更新这个Index的InSyncAllocations配置，将这个Replica从中移除，移除后它就不再承担读请求。在Meta更新到各个Node之前，用户可能还会读到这个Replica的数据，但是更新了Meta之后就不会了。所以这个方案并不是非常的严格，考虑到ES本身就是一个近实时系统，数据写入后需要refresh才可见，所以一般情况下，在短期内读到旧数据应该也是可接受的。

3.2 shard自身角度:

在这里插入图片描述

在每一个Shard中，写入流程分为两部分, 先写入Lucene，再写入TransLog。

写入请求到达Shard后，先写Lucene文件。
此时索引还在内存里面，接着去写TransLog。
写完TransLog后，刷新TransLog数据到磁盘上，并且保留一定的translog中的数据。例如：在进行数据恢复，可以通过translog来进行数据回放，而不是基于数据副本的恢复。提高磁盘的利用率。
写磁盘成功后，请求返回给用户。

（1）为什么引入translog？
当一个文档写入Lucence后是存储在内存中的，即使执行了refresh操作仍然是在文件系统缓存中，如果此时服务器宕机，那么这部分数据将会丢失。为此ES增加了translog，当进行文档写操作时会先将文档写入Lucene，然后写入一份到translog，写入translog是落盘的，这样就可以防止服务器宕机后数据的丢失。由于translog是追加写入，因此性能比较好。

而且key value的形式写Translog， Key是Id, Value是Doc的内容。当查询的时候，如果请求的是GetDocById则可以直接根据_id从translog中获取。满足nosql场景的实时性。

（2）为什么es要先写入lucene，后写入translog？
Lucene的内存写入会有很复杂的逻辑，很容易失败，比如分词，字段长度超过限制等，比较重，为了避免TransLog中有大量无效记录，为了减少写入失败回滚的复杂度和提高速度，所以就把写Lucene放在了最前面。

当一个文档写入Lucene后是不能被立即查询到的，Elasticsearch提供了一个refresh操作，会定时地为内存中新写入的数据生成一个新的segment，此时被处理的文档均可以被检索到。refresh操作的时间间隔由refresh_interval参数控制，默认为1s。

flush操作

另外每30分钟或当translog达到一定大小(由index.translog.flush_threshold_size控制，默认512mb), ES会触发一次flush操作，此时ES会先执行refresh操作将buffer中的数据生成segment，然后调用lucene的commit方法将所有内存中的segment fsync到磁盘。此时lucene中的数据就完成了持久化。

merge操作

在这里插入图片描述

由于refresh默认间隔为1s中，因此会产生大量的小segment，为此ES会运行一个任务检测当前磁盘中的segment，对符合条件的segment进行合并操作，减少lucene中的segment个数，提高查询速度，降低负载。

不仅如此，merge过程也是文档删除和更新操作后，旧的doc真正被删除的时候。用户还可以手动调用_forcemerge API来主动触发merge，以减少集群的segment个数和清理已删除或更新的文档。

当索引的时候，刷新（refresh）操作会创建新的段并将段打开以供搜索使用。
合并进程选择一小部分大小相似的段，并且在后台将它们合并到更大的段中。这并不会中断索引和搜索。
新的段被打开用来搜索
老的段会被删除

在这里插入图片描述

四、更新操作：

在这里插入图片描述

更新流程：

客户端A、B发起Updata操作，并几乎同时获取同一个文档, 一并获得_version版本信息, 假设此时_version=1。
客户端A将版本V1的全量Doc和请求中的部分字段Doc合并为一个完整的Doc。Update请求就变成了Index请求。
Elasticsearch在写入索引时, 检查客户端A提交的文档的版本信息(这里仍然是1) 和现存的文档的版本信息(这里也是1), 发现相同后, 执行写入操作, 并修改版本号_version=2。
客户端B也修改文档中的部分内容, 其操作写回索引的速度稍慢. 此时同样执行过程(4): ES发现客户端B提交的文档的版本为1, 而现存文档的版本为2 ===> 发生冲突, 此次update将失败。
update操作失败后, 将重复(1) - (3) 过程, 并重复几次。参数有配置控制。