快速掌握Elasticsearch检索之二：滚动查询(scrool)获取全量数据(golang)

Elasticsearch8.17.0在mac上的安装

Kibana8.17.0在mac上的安装

Elasticsearch检索方案之一：使用from+size实现分页

1、滚动查询的使用场景

滚动查询区别于上一篇文章介绍的使用from、size分页检索，最大的特点是，它能够检索超过10000条外的所有文档，可以理解为是一种全量检索的技术方案，也正是因为这种特性，使得滚动查询的代价非常高昂，检索过程消耗大量的内存，所以对于实时检索的场景，滚动查询是不适用的。

那滚动查询使用在什么场景呢？主要是应用在离线、检索全量数据，对于实时性要求不高的场景，比如一个数据平台，前台页面展示的数据用来预览，可以使用from+size分页查询，以提升检索效率以及平台的用户体验，如果还需要检索全量数据用于二次使用，那么后台离线检索全量就需要使用滚动查询以获取到全量数据，这将是一个耗费大量资源和时间的过程。

2、使用Kibana直观体验滚动查询

初始化滚动查询：

GET /new_tag_202411/_search?scroll=1m
{"size": 10,"sort":[{"doc_id":{"order": "asc"}}]
}

检索条件设置返回2条数据，按【doc_id】字段升序排列，doc_id分别为1-10的文档。

scroll=1m，表示Elasticsearch允许等待的最长时间是1分钟，如果在一分钟之内，接下来的 scroll 请求没有到达的话，那么当前请求的上下文将会失效：

 从上图返回可以看出，有一个【_scroll_id】字段，这个字段非常重要，接下来的滚动查询需要使用这个字段：

第一次滚动，返回doc_id从11开始的数据，第二次滚动时，需要使用第一次滚动返回的【_scroll_id】替换滚动请求，数据从doc_id为21的数据开始返回，之后循环这个过程，直到检索到全部数据。

注意一点，在测试过程中，我创建了多次滚动查询，发现scrool_id特别像，大家别误以为scrool_id没变，比如以下三个scrool_id，每个id只有3个字符不一样：

FGluY2x1ZGVfY29udGV4dF91dWlkDXF1ZXJ5QW5kRmV0Y2gBFng3akdDTWthVFZLVTE0ODhLdGdaR1EAAAAAAAAWbhZZZEloTnlyU1FGaTgxQV9QR1pXTUdR

FGluY2x1ZGVfY29udGV4dF91dWlkDXF1ZXJ5QW5kRmV0Y2gBFng3akdDTWthVFZLVTE0ODhLdGdaR1EAAAAAAAActhZZZEloTnlyU1FGaTgxQV9QR1pXTUdR

FGluY2x1ZGVfY29udGV4dF91dWlkDXF1ZXJ5QW5kRmV0Y2gBFng3akdDTWthVFZLVTE0ODhLdGdaR1EAAAAAAAAjDxZZZEloTnlyU1FGaTgxQV9QR1pXTUdR

3、代码实现滚动查询(golang)

首先是初始化一个滚动查询：

res, err := client.Search(client.Search.WithIndex("new_tag_202411"),client.Search.WithBody(strings.NewReader(dslQuery.BuildJson())),client.Search.WithScroll(time.Minute*1),
)

这行代码：

client.Search.WithScroll(time.Minute*1)

就是在设置滚动查询上下文的有效时间，其他几行很容易理解。

这几行代码执行完成后，除了能拿到检索数据，还能拿到scroll_id。之后就可以进行滚动查询：

for {docs = Documents{}res, err = client.Scroll(client.Scroll.WithScrollID(scrollId),client.Scroll.WithScroll(time.Minute),)if err != nil {fmt.Println("scroll err:", err.Error())return}err = json.NewDecoder(res.Body).Decode(&docs)if err != nil {fmt.Println("json decode err:", err)return}if len(docs.Hits.Hits) == 0 {break}fmt.Println("search count:", len(docs.Hits.Hits))scrollId = docs.ScrollID
}

这里要注意的一点是，循环滚动时，每个轮次，必须更新scrool_id为上一次滚动返回的值，如上面最后一行代码。

L17-L19行的代码，表示已经查出所有数据，本次没有数据了，同时循环结束。

4、一个必须要考虑的问题

对于滚动查询，前面也说过，会创建一个上下文，当es中存在的上下文数量超过一定限制后，将无法再次创建滚动查询，从而无法检索数据，这个【限制】es默认是500个，我们可以通过es的api查看当前系统中已经创建的上下文数量：

GET /_nodes/stats/indices/search

默认情况下，只要【open_contexts】值小于500，都能正常进行滚动查询，如果已经创建了500个上下文，就会出现问题，下面测试一下，利用代码，创建500个上下文：

 如上图，上下文已经创建500个，运行代码，再次执行滚动查询的动作：

无法查出任何数据，但是以下代码也无任何的报错：

res, err := client.Search(client.Search.WithIndex("new_tag_202411"),client.Search.WithBody(strings.NewReader(dslQuery.BuildJson())),client.Search.WithScroll(time.Minute*100),
)
if err != nil {fmt.Println("search err:", err.Error())return
}

没有走到err分支，经过调试发现，res的结构中的http状态码变了，我们加一行打印：

res, err := client.Search(client.Search.WithIndex("new_tag_202411"),client.Search.WithBody(strings.NewReader(dslQuery.BuildJson())),client.Search.WithScroll(time.Minute*100),)if err != nil {fmt.Println("search err:", err.Error())return}fmt.Println("resp code:", res.StatusCode)err = json.NewDecoder(res.Body).Decode(&docs)if err != nil {fmt.Println("decode err:", err.Error())return}

 运行结果如下：

状态码由正常值0变成了429，所以，在执行滚动查询时，我们需要加上对状态码的判断，以捕获到上下文超限的情况，否则没有检索到数据，还以为系统出bug了呢。

这个问题就是滚动查询的一个短板，系统用户量大了，发起滚动查询一旦超过500，就会影响用户检索数据，当然了，es还是有其他解决方案来进行全量的数据检索，还是那句话，下一篇文章再写。

5、所有代码

github：GitHub - liupengh3c/career 

代码位于以下文件：

https://github.com/liupengh3c/career/blob/main/elastic/scrool/main.go

代码也粘过来吧：

package mainimport ("fmt""os""strings""time""github.com/elastic/go-elasticsearch/v8"jsoniter "github.com/json-iterator/go""github.com/liupengh3c/esbuilder"
)// 最外层数据结构
type Documents struct {ScrollID string      `json:"_scroll_id"`Shards   Shards      `json:"_shards"`Hits     HitOutLayer `json:"hits"`TimedOut bool        `json:"timed_out"`Took     int         `json:"took"`
}
type Shards struct {Failed     int `json:"failed"`Skipped    int `json:"skipped"`Successful int `json:"successful"`Total      int `json:"total"`
}
type HitOutLayer struct {Hits     []Hits  `json:"hits"`MaxScore float64 `json:"max_score"`Total    Total   `json:"total"`
}
type Hits struct {ID     string         `json:"_id"`Index  string         `json:"_index"`Score  float64        `json:"_score"`Source map[string]any `json:"_source"`Type   string         `json:"_type"`
}
type Total struct {Relation string `json:"relation"`Value    int    `json:"value"`
}func main() {client, err := NewEsClient()if err != nil {fmt.Println("create client err:", err.Error())return}fmt.Println("connect success")for i := 0; i < 510; i++ {ScrollSearch(client)}
}
func NewEsClient() (*elasticsearch.Client, error) {cert, _ := os.ReadFile("/Users/liupeng/Documents/study/elasticsearch-8.17.0/config/certs/http_ca.crt")client, err := elasticsearch.NewClient(elasticsearch.Config{Username:  "elastic",Password:  "XBS=adqa799j_Aoz=A+h",Addresses: []string{"https://127.0.0.1:9200"},CACert:    cert,})if err != nil {// fmt.Println("create client err:", err.Error())return client, err}return client, nil
}func ScrollSearch(client *elasticsearch.Client) {var json = jsoniter.ConfigCompatibleWithStandardLibrarydocs := Documents{}dslQuery := esbuilder.NewDsl()boolQuery := esbuilder.NewBoolQuery()dslQuery.SetOrder(esbuilder.NewSortQuery("doc_id", "asc"))dslQuery.SetQuery(boolQuery)dslQuery.SetSize(10000)res, err := client.Search(client.Search.WithIndex("new_tag_202411"),client.Search.WithBody(strings.NewReader(dslQuery.BuildJson())),client.Search.WithScroll(time.Minute*20),)if err != nil {fmt.Println("search err:", err.Error())return}err = json.NewDecoder(res.Body).Decode(&docs)if err != nil {fmt.Println("decode err:", err.Error())return}fmt.Println("search count:", len(docs.Hits.Hits))scrollId := docs.ScrollIDfor {docs = Documents{}res, err = client.Scroll(client.Scroll.WithScrollID(scrollId),client.Scroll.WithScroll(time.Minute),)if err != nil {fmt.Println("scroll err:", err.Error())return}err = json.NewDecoder(res.Body).Decode(&docs)if err != nil {fmt.Println("decode err:", err.Error())return}defer res.Body.Close()if res.StatusCode == 429 {fmt.Println("scroll contexts is more than 500")return}if len(docs.Hits.Hits) == 0 {break}fmt.Println("search count:", len(docs.Hits.Hits))scrollId = docs.ScrollID}client.ClearScroll(client.ClearScroll.WithScrollID(scrollId),)
}