使用Go操作Elasticsearch切片滚动遇数据异常问题求助

我使用go-elasticsearch/v7库做滚动查询，普通滚动能正常拿到432734条结果，但耗时超1分钟，因此尝试通过切片滚动并发提速。

普通滚动实现代码 #

import es "github.com/elastic/go-elasticsearch/v7"
[...]

type Query struct {
    Size  int          `json:"size"`
    Slice *QuerySlice  `json:"slice,omitempty"`
    Aggs  *Aggs        `json:"aggs,omitempty"`
    Query *QueryFilter `json:"query,omitempty"`
    Sort  []string     `json:"sort,omitempty"`
}

type QuerySlice struct {
    ID  int `json:"id"`
    Max int `json:"max"`
}

[...]

type QueryFilter struct {
    Bool QFBool `json:"bool"`
}

type QFBool struct {
    Filter Filter `json:"filter"`
}

type Filter []map[string]map[string]interface{}

func (q *Query) AsBody() (*bytes.Reader, error) {
    queryBytes, err := json.Marshal(q)
    if err != nil {
        return nil, err
    }

    return bytes.NewReader(queryBytes), nil
}

[...]

type Result struct {
    ScrollID     string `json:"_scroll_id"`
    Took         int
    TimedOut     bool   `json:"timed_out"`
    HitSet       HitSet `json:"hits"`
    Aggregations Aggregations
}

type HitSet struct {
    Total struct {
        Value int
    }
    Hits []Hit
}

[...]

func main() {
    query := &Query{
        Size:  MaxSize,
        Sort:  []string{"_doc"},
        Query: &QueryFilter{Bool: QFBool{Filter: filter}},
    }

    qbody, err := query.AsBody()

    resp, err := client.Search(
        client.Search.WithIndex(index),
        client.Search.WithBody(qbody),
        client.Search.WithSize(maxSize),
        client.Search.WithScroll(scrollTime),
    )

    result, err := parseResponse(resp)
}

切片滚动实现代码 #

total := 0

for i := 0; i < maxSlices; i++ {
    query := &Query{
        Size: maxSize,
        Sort: []string{"_doc"},
        Slice: &QuerySlice{
            ID:  i,
            Max: maxSlices,
        },
        Query: &QueryFilter{Bool: QFBool{Filter: filter}},
    }

    qbody, err := query.AsBody()
    if err != nil {
        return nil, err
    }

    resp, err := client.Search(
        client.Search.WithIndex(index),
        client.Search.WithBody(qbody),
        client.Search.WithSize(maxSize),
        client.Search.WithScroll(scrollTime),
    )
    if err != nil {
        return nil, err
    }

    result, err := parseResponse(resp)
    if err != nil {
        return nil, err
    }

    fmt.Printf("slice total: %d; hits: %d\n", result.HitSet.Total.Value, len(result.HitSet.Hits))

    total += result.HitSet.Total.Value
}

fmt.Printf("overall total: %d\n", total)

测试结果 #

• maxSlices=2时，结果符合预期：

slice total: 194104; hits: 10000
slice total: 238630; hits: 10000
overall total: 432734

• maxSlices=3时，总数远小于预期：

slice total: 125374; hits: 10000
slice total: 80754; hits: 10000
slice total: 125374; hits: 10000
overall total: 331502

• maxSlices=6时，部分切片无数据：

slice total: 10117; hits: 10000
slice total: 11253; hits: 10000
slice total: 114486; hits: 10000
slice total: 0; hits: 0
slice total: 0; hits: 0
slice total: 0; hits: 0
overall total: 135856

结果不稳定，比如maxSlices=3有时正常有时异常，但普通滚动每日结果一致。请问我是否误解或误用了Elasticsearch切片滚动？

问题原因与解决方法 #

核心问题：未完成切片的完整滚动流程 #

你当前的切片滚动代码只执行了每个切片的第一次Search请求，仅拿到第一页数据就停止了，没有继续调用Scroll接口拉取该切片的剩余数据。另外，你依赖的result.HitSet.Total.Value在切片场景下本身就可能存在偏差——Elasticsearch切片基于文档_uid哈希分片，当切片数超过索引分片数时，哈希分布易出现不均，部分切片会分配不到数据；同时滚动查询的Total值是近似统计，在有实时数据变更的场景下准确性无法保证。

正确的切片滚动实现逻辑 #

1. 对每个切片，先执行初始Search获取scroll_id和第一页数据；
2. 循环调用client.Scroll接口，传入scroll_id，直到返回的hits为空；
3. 累加每个切片实际获取到的hits数量，而非依赖返回的Total.Value。

修正后的代码示例 #

var wg sync.WaitGroup
var totalHits int64
var mutex sync.Mutex

maxSlices := 3
scrollTime := "1m"
maxSize := 10000

for i := 0; i < maxSlices; i++ {
    wg.Add(1)
    go func(sliceID int) {
        defer wg.Done()
        sliceTotal := 0

        // 初始Search创建切片滚动上下文
        query := &Query{
            Size: maxSize,
            Sort: []string{"_doc"},
            Slice: &QuerySlice{
                ID:  sliceID,
                Max: maxSlices,
            },
            Query: &QueryFilter{Bool: QFBool{Filter: filter}},
        }
        qbody, err := query.AsBody()
        if err != nil {
            fmt.Printf("slice %d: 构建查询失败: %v\n", sliceID, err)
            return
        }

        resp, err := client.Search(
            client.Search.WithIndex(index),
            client.Search.WithBody(qbody),
            client.Search.WithSize(maxSize),
            client.Search.WithScroll(scrollTime),
        )
        if err != nil {
            fmt.Printf("slice %d: 初始查询失败: %v\n", sliceID, err)
            return
        }
        defer resp.Body.Close()

        result, err := parseResponse(resp)
        if err != nil {
            fmt.Printf("slice %d: 解析初始响应失败: %v\n", sliceID, err)
            return
        }
        sliceTotal += len(result.HitSet.Hits)
        scrollID := result.ScrollID

        // 循环滚动获取剩余数据
        for {
            resp, err := client.Scroll(
                client.Scroll.WithScrollID(scrollID),
                client.Scroll.WithScroll(scrollTime),
            )
            if err != nil {
                fmt.Printf("slice %d: 滚动查询失败: %v\n", sliceID, err)
                break
            }
            defer resp.Body.Close()

            result, err := parseResponse(resp)
            if err != nil {
                fmt.Printf("slice %d: 解析滚动响应失败: %v\n", sliceID, err)
                break
            }

            hitsLen := len(result.HitSet.Hits)
            if hitsLen == 0 {
                break
            }
            sliceTotal += hitsLen
            scrollID = result.ScrollID
        }

        // 并发安全累加总命中数
        mutex.Lock()
        totalHits += int64(sliceTotal)
        mutex.Unlock()
        fmt.Printf("slice %d: 实际获取命中数: %d\n", sliceID, sliceTotal)
    }(i)
}

wg.Wait()
fmt.Printf("总实际获取命中数: %d\n", totalHits)

额外注意事项 #

• 切片数选择：建议将maxSlices设置为索引分片数或其倍数，保证哈希分布均匀，减少空切片出现概率；
• 并发控制：不要设置过大的切片数，避免给Elasticsearch集群造成过载；
• 放弃依赖Total.Value：滚动查询的Total值是近似统计，实际获取的hits数量才是准确值；
• 合理设置scroll超时：超时时间需足够覆盖单切片的滚动周期，但也不要过长浪费集群资源。

内容的提问来源于stack exchange，提问作者sbs