Elasticsearch建立索引流程

前言

说明：本文章使用的ES版本是：6.7.0

在上一篇文章搜索引擎ElasticSearch的启动过程中，介绍了ES的启动过程。

由此可知，在ES启动过程当中，建立Node对象（new Node(environment)）时，初始化了RestHandler，由其名字能够知道这是用来处理Rest请求的。

在ES源码中，RestHandlerAction以下图：

其中：

• admin

  • cluster：处理集群相关请求
  • indices：处理索引相关请求
• cat：平常查询
• document：文档处理
• ingest：pipeline处理。pipeline？干吗的
• search：搜索

接下来咱们具体的看一下ES是如何建立索引的：org.elasticsearch.rest.action.document.RestIndexAction

数据概念和结构

一个完整的ES集群由如下几个基本元素组成

名称	概念	对应关系型数据库概念	说明
Cluster	集群		一个或多个节点的集合，经过启动时指定名字做为惟一标识，默认cluster-state
node	节点		启动的ES的单个实例，保存数据并具备索引和搜索的能力，经过名字惟一标识，默认node-n
index	索引	Database	具备类似特色的文档的集合，能够对应为关系型数据库中的数据库，经过名字在集群内惟一标识
type	文档类别	Table	索引内部的逻辑分类，能够对应为Mysql中的表，ES 6.x 版本中，一个索引只容许一个type，再也不支持多个type。7.x版本中，type将废弃。
document	文档	Row	构成索引的最小单元，属于一个索引的某个类别，从属关系为： Index -> Type -> Document，经过id 在Type 内惟一标识
field	字段	Column	构成文档的单元
mapping	索引映射（约束）	Schema	用来约束文档字段的类型，能够理解为索引内部结构
shard	分片		将索引分为多个块，每块叫作一个分片。索引定义时须要指定分片数且不能更改，默认一个索引有5个分片，每一个分片都是一个功能完整的Index，分片带来规模上（数据水平切分）和性能上（并行执行）的提高，是ES数据存储的最小单位
replicas	分片的备份		每一个分片默认一个备份分片，它能够提高节点的可用性，同时可以提高搜索时的并发性能（搜索能够在所有分片上并行执行）

一个ES集群的结构以下：

每一个节点默认有5个分片，每一个分片有一个备分片。

6.x版本以前的索引的内部结构：

说明：ES 6.x 版本中，相同索引只容许一个type，再也不支持多个type。7.x版本中，type将废弃。

因此，6.x版本的索引结构以下：

7.x版本的索引结构以下：

索引一个文档

启动ES实例后，发送以下请求：

curl -X PUT 'localhost:9200/index_name/type_name/1' -H 'Content-Type: application/json' -d '
{
  "title": "我是文件标题，可被搜索到",
  "text":  "文本内容，ES时如何索引一个文档的",
  "date":  "2019/01/01"
}'

其中:

• index_name:表示索引名称
• type_name:类别名称
• 1：文档ID

ES执行流程：

客户端：

1. BaseRestHandler#handleRequest：处理请求

2. RestIndexAction#prepareRequest：封装request，识别行为，容许的行为以下，默认INDEX

   enum OpType {
       // Index the source. If there an existing document with the id, it will be replaced.
       INDEX(0),
       // Creates the resource. Simply adds it to the index, if there is an existing document with the id, then it won't be removed.
       CREATE(1),
       /** Updates a document */
       UPDATE(2),
       /** Deletes a document */
       DELETE(3);
       
       ...
   }

3. 参数检查，查看是否有关键字，并获取相关关键字的值

   0 = "parent"
   1 = "pretty"
   2 = "version_type"
   3 = "format"
   4 = "index"
   5 = "refresh"
   6 = "error_trace"
   7 = "type"
   8 = "timeout"
   9 = "pipeline"
   10 = "routing"
   11 = "if_seq_no"
   12 = "if_primary_term"
   13 = "wait_for_active_shards"
   14 = "id"
   15 = "op_type"
   16 = "human"
   17 = "filter_path"

4. NodeClient#doExecute：指定执行该请求的actionName：indices:data/write/index

5. TransportAction#execute()：将请求封装成CreateIndexRequest并发送到服务端，处理发送前置任务

   1. IndexRequest#validate：校验参数内容，type、source、contentType

   2. 这里若是是更新或者删除操做，检查是否传入ID字段，没传如则报错

      if (opType() != OpType.INDEX && id == null) {
          addValidationError("an id is required for a " + opType() + " operation", validationException);
      }

   3. 判断ID长度，最长不能超过512个字符

Transport层

Transport将request封装成Task，将请求发送给服务端

服务端

1. 服务端根据actionName获取具体响应请求的action，此处为执行：TransportBulkAction#doExecute()

2. 读取AutoCreateIndex#AUTO_CREATE_INDEX_SETTING，该值由配置文件elasticsearch.yml中的auto_create_index控制，true表示当插入的索引不存在时，自动建立该索引
   

   1. 若是"auto_create_index"为true：

      1. 分析bulkRequest中的全部请求中的全部index，生成Set<String> indices，

      2. 而后遍历indices，判断索引名称是否存在

         1. 索引不存在：将请求转发给TransportCreateIndexAction#masterOperation，建立索引，且索引建立完成后，执行第2步
         2. 索引存在：启动异步进程BulkOperation，该进程将负责建立索引
   2. 若是"auto_create_index"为false，则索引不存在的写入文档的请求

TransportCreateIndexAction 建立索引过程

1. 该类继承TransportMasterNodeAction，它会启动一个异步线程来执行任务，若是当前节点是master节点，则执行masterOperation，不然转发给master节点(每一个节点在启动时会加入集群，同时保存完整的集群信息，该信息又Discovery模块维护)
2. TransportCreateIndexAction将CreateIndexRequest转换为CreateIndexClusterStateUpdateRequest，将请求做为参数，调用MetaDataCreateIndexService#createIndex
3. 调用MetaDataCreateIndexService#onlyCreateIndex，该方法负责在clusterstate中建立新的index，而且等待指定数目（默认为1）状态为active的分片副本建立完成（activeShardsObserver.waitForActiveShards方法实现），最终返回给listener。

4. onlyCreateIndex方法，其内部执行clusterService.submitStateUpdateTask，提交集群状态修改任务，提交任务的执行逻辑是AckedClusterStateUpdateTask类内部的execute方法。其内部逻辑为：

   1. 校验index的名字和settings是否合法（好比index名不能有大写，若是有别名，判断是否有重名）
   2. 根据index name 查找合适的模板信息，即mapping
   3. 构建indexSettingsBuilder，能够认为是该索引的默认环境变量
   4. 准备工做完成，开始写入索引IndicesService#createIndex，写入索引的动做由IndexModule#newIndexService完成
   5. 为indicesService服务增长index服务，mapperService服务，同时合并新老mappings
   6. 构建IndexMetaData，并生成新的ClusterState
   7. 若是index状态open，执行allocationService.reroute将分片分配到其余节点
   8. 最后删除索引服务（indicesService.removeIndex）

5. 上一步修改完成clusterstate后

   1. 若是是master节点同步集群状态（若是是master）
   2. 通知集群状态监听器listener，其余节点接收到集群状态变化，启动indicesService服务

BulkOperation 写入文档过程

1. 获取最新的集群状态clusterstate

2. 遍历request中的文档

   1. 获取文档操做类型OpType，写入文档
   2. 对文档作一些加工，主要包括：解析routing(若是mapping里有的话)、指定的timestamp(若是没有带timestamp会使用当前时间)，若是文档没有指定id字段，会自动生成一个base64UUID做为id字段
      
3. 再次遍历全部的request，获取获取每一个request应该发送到的shardId，获取的过程是这样的：若是上一步获取到了routing则取routing，不然取文档ID，取其hash值（哈希算法 Murmur3Hash）而后对当前索引的分片数量取模，获得分片ID：shardId
4. 将相同分片的请求分组，将请求封装成BulkShardRequest，经过TransportBulkAction将请求发送到分片所在节点
5. 请求转发到Node节点更新主分片，TransportReplicationAction.execute(),建立一个ReroutePhase异步线程，并执行，此处文档会写入主分片buffer中（InternalEngine#indexIntoLucene），最后并启动异步进程ReplicationPhase，更新副分片
6. 至此，文档写入完成，但只是将数据写入内存buffer和transLog中，以后还有异步进程将数据refresh到索引中使其可搜索，将数据flush到磁盘

文档写入总结

1. 经过副本分片和Translog日志保障数据安全和一致性
2. 在可用性和一致性二者的取舍中，ES更看重可用性。主分片写入后，便可搜索。所以若是请求落到副分片可能出现不一致的状况，可是在搜索业务中，这种短期的不一致大可能是能够接受的

系列文章

1. 搜索引擎ElasticSearch源码编译和Debug环境搭建
2. 搜索引擎ElasticSearch的启动过程
3. Elasticsearch建立索引流程
4. Elasticsearch搜索过程详解
5. Elasticsearch搜索相关性排序算法详解
6. Elasticsearch中的倒排索引