【ES三周年】ES查询—海量数据搜索深度分页优化

原创

后面牵个人

修改于 2023-04-29 16:11:50

4.7K18

文章被收录于专栏：ES搜索ES搜索

背景

最近在实际项目中查询条件上越来越复杂，mysql的筛选已无法支撑，准备将所有搜索筛选改为es查询。鉴于这种情况，本文调研了from-size，search_after，scroll api, search_after (PIT) 这四种查询优劣。

From - size 普通分页

使用 from + size 对翻页

from 未指定，默认值是 0，定义了需要跳过的 hits 数，默认 0。
size 未指定，默认值是 10，定义了需要返回的 hits 数目的最大值。

该方案使用简单，在翻页数目较多即 from 较大或者 size 特别大的情况，会存在深翻页问题。ES 默认认的单页查询最大限制max_result_window 为10000 。

深翻页问题原因：ES 本身采用了分布式的架构，在存储数据时，会将其分配到不同的 shard 中。在查询时，如果 from 值过大，就会导致分页起点太深。每个 shard 查询时，都会将 from 之前位置的所有数据和请求 size 的总数返回给coordinator。对于coordinator来说，会显著导致内存和CPU使用率升高，特别是在高并发的场景下，导致性能下降或者节点故障。

例如ES 共有 4 个shard，并且每个shard没有副本。假如分页的大小为 10，想取第11 页的内容。则对应的 from = 100，size = 10。

ES 的查询过程为：

每个shard将所在数据加载到内存并排序，然后取前 110 个，返回给coordinator。
每个shard都执行上面的操作。
最后coordinator将 110 * 4 = 440 条数据排序，然后取 10 条数据返回。

可以发现，from的位置太深，必然产生如下问题：

返回给coordinator数值太大，实际需要 10条数据，但却给coordinator440 条数据
coordinator需要处理每个shard返回前 11页的结果。但需要的仅是第 11 页的内容，却对前 44 页的内容进行了排序，浪费了内存和 cpu 的资源。

优点

实现较为简单。
可以指定任意合理的页码，实现跳页查询。

缺点

查询分页受限于max_result_window设置，不能无限制翻页。
查询分页性能不稳定，越往后翻页越慢，存在深度翻页问题。

go代码

func SearchTaskSampleByQuery(ctx context.Context, query *elastic.BoolQuery, page *common_base.PageReq) (int64, []es.TaskSample, error) {
   var taskSampleList []es.TaskSample
   esClient, err := olivere7.Get(config.EsNameSrv)
   if err != nil {
      return 0, taskSampleList, constant.Errorf(constant.CodeErrEsSearchFail, "SearchTakSample get es client error: %w", err)
   }
   search := esClient.Search().Index(索引名)
   search.Query(elastic.NewBoolQuery().Must(query))
   // 排序
   search.Sort("id", false)
   // 分页
   if page != nil {
      from := page.PageSize * (page.PageNum - 1)
      search.From(str.StringToInt(fmt.Sprint(from)))
      search.Size(str.StringToInt(fmt.Sprint(page.PageSize)))
   }
   search.TrackTotalHits(true)
   // 执行
   esResult, err := search.Pretty(true).Do(ctx)
   if err != nil {
      return 0, taskSampleList, constant.Errorf(constant.CodeErrEsSearchFail, "SearchTaskSampleByQuery Do error: %w", err)
   }
   // es查询数据转化
   return parseEsSearchTaskSampleResult(ctx, esResult)
}

Scroll 查询分页

可以有效地执行大批量的文档查询。游标查询会取某个时间点的快照数据。查询初始化之后索引上的任何变化会被它忽略。它通过保存旧的数据文件来实现这个特性，结果就像保留初始化时的索引视图一样。

这个分页的用法，不是为了实时查询数据，而是为了一次性查询大量的数据（甚至是全量数据）。

具体使用方法：

第一次查询时，会生成一个 scrollId ，并将所有符合搜索条件的搜索结果缓存起来。注意，这里只是缓存的 doc_id ，并不是真的缓存了所有的文档数据，取数据是在 fetch 阶段完成的。
后续查询时，需要携带上一次查询返回的 scrollId和scrolles把本次快照(search context)的结果缓存起来的有效时间。

ES的检索分为查询（query）和获取（fetch）两个阶段，query阶段比较高效，只是查询满足条件的文档id汇总起来。fetch阶段则基于每个分片的结果在coordinating节点上进行全局排序，然后最终计算出结果。

Scroll查询只搜索到了所有的符合条件的 doc_id (官方推荐用 doc_id 进行排序，因为本身缓存的就是 doc_id ，如果用其他字段排序会增加查询量)，并将它们排序后保存在search context，但是并没有将所有数据进行fetch，而是每次scroll，读取size个文档，并返回此次读取的后一个文档以及上下文状态，用以告知下一次需要从哪个shard的哪个文档之后开始读取。

优点

依据数据快照查找，能有效保障数据的一致性。
查询不受限于 index.max_result_window 影响。

缺点

查询结果非实时，对于数据的变更不会反映到快照上。
维护 scroll_id 和历史快照，并且，还必须保证 scroll_id 的存活时间，这对服务器是一个巨大的负荷。

go代码

// ScrollTaskSample scroll离线查询
func ScrollTaskSample(ctx context.Context, query *elastic.BoolQuery, scrollId, scrollTime string) (int64, []es.Sample, error) {
   var sampleList []es.Sample
   esClient, err := olivere7.Get(config.EsNameSrv)
   if err != nil {
      return 0, sampleList, constant.Errorf(constant.CodeErrEsSearchFail, "SearchSampleByQuery get es client error: %w", err)
   }
   search := esClient.Scroll().Index(SampleIndex)
   search.Query(elastic.NewBoolQuery().Must(query))
   // 排序
   search.Sort("id", false)
   search.Size(1000)
   // 7.0版本命中数
   search.TrackTotalHits(true)
   // 第一次调用，传入scrollId空字符串, scrollTime 5s, 获取 esResult.ScrollId
   // 后续调用，传入 esResult.ScrollId, 5m, 直到命中数组长度为0即可
   search.ScrollId(scrollId)
   // 快照时间
   search.Scroll(scrollTime)
   // 执行
   esResult, err := search.Pretty(true).Do(ctx)
   if err != nil {
      return 0, sampleList, constant.Errorf(constant.CodeErrEsSearchFail, "SearchSampleByQuery Do error: %w", err)
   }
   return parseEsSearchSampleResult(ctx, esResult)
}

Search_after 分页

search_after 查询: 使用上次查询的最后一条数据来进行下一次查询。因为每一页的数据都是依赖于上一页最后一条数据，所以无法跳页请求。

具体使用方法：

第一次请求时，会返回一个包含 sort 排序值的数组
在下一次请求时，可以将前面一次请求返回结果中 sort 排序值用于入参，以便抓取下一页的数据

例如ES 共有 4 个shard，并且每个shard没有副本。假如分页的大小为 10，想取第11 页的内容。对应的 from = 100，size = 10.

ES 的查询过程为：

每个shard根据sort游标，拿出满足条件的10个样本，返回给coordinator.
每个shard都执行上面的操作。
最后coordinator将 10 * 4 = 40 条数据排序，然后取 10 条数据返回。

优点

无状态查询，可以防止在查询过程中，数据的变更无法及时反映到查询中。
不需要维护 scroll_id ，不需要维护快照，因此可以避免消耗大量的资源。
查询不受限于 index.max_result_window 影响。

缺点

由于无状态查询，因此在查询期间的变更可能会导致跨页面的不一致。
排序顺序可能会在执行期间发生变化，具体取决于索引的更新和删除。
至少需要制定一个的不重复字段来排序。
它不适用于大幅度跳页查询，或者全量导出，对第N页的跳转查询相当于对es不断重复的执行N次search after，而全量导出则是在短时间内执行大量的重复查询。

go代码

// SearchAfterTaskSample 游标查询分页
func SearchAfterTaskSample(ctx context.Context, query \*elastic.BoolQuery, sortFlag \*[]interface{}) (int64, []es.Sample, error) {
   var sampleList []es.Sample
   esClient, err := olivere7.Get(config.EsNameSrv)
   if err != nil {
      return 0, sampleList, constant.Errorf(constant.CodeErrEsSearchFail, "SearchSampleByQuery get es client error: %w", err)
   }
   search := esClient.Search()
   search.Query(elastic.NewBoolQuery().Must(query))
   // 排序
   search.Sort("id", false)
   if sortFlag != nil {
      // search\_after
      search.SearchAfter(sortFlag...)
   }
   // 7.0版本命中数
   search.TrackTotalHits(true)
   // 执行
   esResult, err := search.Pretty(true).Do(ctx)
   if err != nil {
      return 0, sampleList, constant.Errorf(constant.CodeErrEsSearchFail, "SearchSampleByQuery Do error: %w", err)
   }
   // 下一页游标 sortFlag := &esResult.Hits.Hitslen(esResult.Hits.Hits)-1.Sort
   return parseEsSearchSampleResult(ctx, esResult)
}

Search_after (PIT)分页!

在 7.10以后版本中，ES官方不再推荐使用Scroll方法来进行深分页，而是推荐使用带PIT的 search_after 来进行查询。

PIT可以被看为存储索引数据状态的轻量级视图。创建一个时间点 Point In Time（PIT）保障搜索过程中保留特定事件点的索引状态。有了 PIT，search_after 的后续查询都是基于 PIT 视图进行，能有效保障数据的一致性。

优点

查询分页效果和Scroll完全一致，但平均查询效率提升了30%。引用文章：Elasticsearch Scroll API vs Search After with PIT

相比scroll，内存也得到了优化，es 的查询简化流程：

第一步.用户发送查询dsl
第二步.ES获取shard 内存引用（实际上是ReaderContext 对象引用 ,指向shard的segment 某个状态的数据）
第三步.ES从shard 根据dsl 查询出result
scroll 的原理是直接执行第一、二、三步，然后在内存缓存result 的全部结果同时构建一个游标，然后通过游标移动逐步返回结果
pit 的原理是先执行第二步，并且缓存了shard 内存引用，后续再做第一步跟第三步，用户后续通过发送dsl并且指定search after 来逐步拉取数据

可以看到pit 的缓存复用率是比scroll 要高的：比如有100w 个scroll 查询进来，内存中需要缓存100w个scroll 查询结果。但是pit 缓存的shard 内存引用实际上很多查询之间是可以复用的，按理说内存消耗会更低。

适合于大批量的拉取数据，非实时处理数据的情况，数据迁移或者索引变更等场景。

缺点

查询无法反应数据的实时性，生成的数据历史快照，对于数据的变更不会反映到快照上。

go代码

// CreatePit 创建时间点
func CreatePit(ctx context.Context, indexName, aliveTime string) (string, error) {
   esClient, err := olivere7.Get(config.EsNameSrv)
   if err != nil {
      return "", constant.Errorf(constant.CodeErrEsSearchFail, "get es client error: %w", err)
   }
   openRsp, err := esClient.OpenPointInTime(indexName).KeepAlive(aliveTime).Pretty(true).Do(ctx)
   if err != nil {
      return "", err
   }
   return openRsp.Id, nil
}

// ClosePit 关闭时间点
func ClosePit(ctx context.Context, pit string) (bool, error) {
   esClient, err := olivere7.Get(config.EsNameSrv)
   if err != nil {
      return false, constant.Errorf(constant.CodeErrEsSearchFail, "get es client error: %w", err)
   }
   closeResp, err := esClient.ClosePointInTime(pit).Pretty(true).Do(ctx)
   if err != nil {
      return false, err
   }
   return closeResp.Succeeded, nil
}

// SearchAfter 游标查询
func SearchAfter(ctx context.Context, query *elastic.BoolQuery, sortFlag []interface{},
   pit, aliveTime string) (*elastic.SearchResult, error) {
   esClient, err := olivere7.Get(config.EsNameSrv)
   if err != nil {
      return nil, constant.Errorf(constant.CodeErrEsSearchFail, "get es client error: %w", err)
   }
   search := esClient.Search()
   // 每次拉取大小设置
   search.Size(10)
   search.Query(elastic.NewBoolQuery().Must(query))
   // 排序
   search.Sort("_shard_doc", true)
   // 时间点传递
   if len(pit) > 0 {
      pointTime := &amp;elastic.PointInTime{
         Id:        pit,
         KeepAlive: aliveTime,
      }
      search.PointInTime(pointTime)
   }
   // 游标传递
   if len(sortFlag) != 0 {
      // search_after
      search.SearchAfter(sortFlag...)
   }
   // 执行
   esResult, err := search.Pretty(true).Do(ctx)
   if err != nil {
      return nil, constant.Errorf(constant.CodeErrEsSearchFail, "SearchAfter Do error: %w", err.Error())
   }
   return esResult, nil
}