当前位置： 首页 > article >正文

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

article 2025/1/5 6:28:30

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 main

import (
	"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.ConfigCompatibleWithStandardLibrary
	docs := 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.ScrollID
	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("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),
	)
}


http://www.kler.cn/a/463425.html

相关文章：

  • 【新教程】华为昇腾NPU的pytorch环境搭建
  • ESP32 I2S音频总线学习笔记（一）：初识I2S通信与配置基础
  • w~多模态~合集1
  • 2024/12/29 黄冈师范学院计算机学院网络工程《路由期末复习作业一》
  • JavaScript的diff库详解（示例：vue项目实现两段字符串比对标黄功能）
  • 【JVM】总结篇-字节码篇
  • Dockerfile 构建继承父镜像的 ENTRYPOINT 和 CMD
  • Python性能分析深度解析：从`cProfile`到`line_profiler`的优化之路
  • 数据结构：排序
  • .NET在中国的就业前景：开源与跨平台带来的新机遇
  • dbN小波构造与求解实例分析-附Matlab代码
  • 数据的简单处理——pandas模块——数据结构（Series和DataFrame对象）
  • 韩国首尔阿里云200M不限流量轻量云主机测试报告
  • Flink源码解析之：如何根据StreamGraph生成JobGraph
  • IP寻址映射与网络通信互联
  • [react] 纯组件优化子
  • JMeter脚本参数化与并发策略
  • Vue 针对浏览器参数过长实现浏览器参数加密解密
  • 人工智能机器学习基础篇】——深入详解强化学习 基础知识，理解马尔可夫决策过程（MDP）、策略、价值函数等关键概念
  • FPGA实现HDMI输出
  • 2024树莓派apt源（可用）
  • 【Leetcode 热题 100】79. 单词搜索
  • Amazon Bedrock 实践 - 利用 Llama 3.2 模型分析全球糖尿病趋势
  • uni-app开发-识图小程序-分类识别功能
  • [微服务] - MQ高级
  • 游戏引擎学习第69天