一、前言：

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

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

Luence 存在的问题：

1. 没有并发设计
   lucene只是一个搜索引擎库，并没有涉及到分布式相关的设计，因此要想使用Lucene来处理海量数据，并利用分布式的能力，就必须在其之上进行分布式的相关设计。

2. 非实时
   将文件写入lucence后并不能立即被检索，需要等待lucene生成一个完整的segment才能被检索

3. 数据存储不可靠
   写入lucene的数据不会立即被持久化到磁盘，如果服务器宕机，那存储在内存中的数据将会丢失

4. 不支持部分更新
   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

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

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

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

（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个数和清理已删除或更新的文档。

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

四、更新操作：

更新流程：

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

五、扩展

为了对比学习，也对比了一下 ES 和我之前学习过的组件一些大方向上的原理做了个比对。

1. HBase VS ES

相同点：

1. 每隔一段比较长的时间/ 日志文件达到一定的大小/ 手动flush，会把内存中数据刷新到磁盘上。
2. 删除数据并不会真正的删除，当发生合并时才会真正的删除数据。
3. hbase 和 es 都会保留一定的translog。

不同点:

1. HBase是先写入日志，然后再写内存，而Elasticsearch是先写内存，最后才写TransLog。
2. hbase只有发生major compact才会真正的删除数据。

2. Kafka vs ES

相似点：

1. ES 是一个索引通过shard来进行水平拆分，Kafka是通过partition来进行水平拆分。
2. ES和Kafka的可靠性都是通过副本来保障。
3. 都会维护一个ISR信息。

不同点：

1. ES 支持读分离 可以在副本分片上读取数据，而Kafka不支持读写分离，读写都必须leader 分区上。

为什么Kafka不支持读写分离，而ES支持读分离？

1. 数据一致性问题。数据从主节点转到从节点必然会有一个延时的时间窗口，这个时间窗口会导致主从节点之间的数据不一致。
2. 延时问题。数据从写入主节点到同步至从节点中的过程需要经历网络→主节点内存→主节点磁盘→网络→从节点内存→从节点磁盘这几个阶段。对延时敏感的应用而言，主写从读的功能并不太适用。

六、读原理

Elasticsearch中每个Shard都会有多个Replica，主要是为了保证数据可靠性，除此之外，还可以增加读能力，因为写的时候虽然要写大部分Replica Shard，但是查询的时候只需要查询Primary和Replica中的任何一个就可以了。

所有的搜索系统一般都是两阶段查询，第一阶段查询到匹配的DocID，第二阶段再查询DocID对应的完整文档，这种在Elasticsearch中称为query_then_fetch。

Query流程：

1. 客户端发送一个 search 请求到 Node 3。
2. Node 3 将查询请求转发到索引的每个主分片或副本分片中。
3. 每个分片返回各自优先队列中所有文档的 ID 和排序值给协调节点，也就是 Node 3
4. 协调节点合并这些值到自己的优先队列中来产生一个全局排序后的结果列表。

Fetch阶段：

查询阶段标识哪些文档满足搜索请求，然后需要取回这些文档。

1. 协调节点首先决定哪些文档“确实”需要被取回，例如：如果查询指定了{"from":90, "size": 10},则只有从第91个开始的10个结果需要被取回。
2. 协调节点向相关node发送GET请求。
3. 分片所在的节点向协调节点返回结果。
4. 协调节点等待所有文档被取得，然后返回给客户端。