Elasticsearch如何对降序排序后的指定数量文档执行sum聚合

你之前的查询拿不到正确结果，核心是三个认知误区：

• 根节点的sort配置只控制hits数组返回的文档顺序，对聚合逻辑完全无效，聚合默认会扫描所有匹配query条件的全量文档计算
• sampler聚合的shard_size是单分片维度的抽样数，假设你的索引有5个主分片，最终参与聚合的是5*30=150条文档，不是全局的30条，结果自然不准。仅当索引只有1个主分片时这个写法能得到正确结果，但生产环境ES索引基本都是多分片架构，没有通用性
• top_hits、普通terms排序都只是返回排序后的文档明细，不支持基于返回的top N结果再做sum二次聚合，没法直接得到你要的求和值

方案1：业务侧计算（最稳妥、性能最优、无版本兼容问题） #

不需要写复杂聚合，直接查询符合时间范围的文档，按create_datetime降序排序后取前30条，拿到duration_seconds字段后在业务代码里循环累加即可，DSL如下：

{
  "size": 30,
  "_source": ["duration_seconds"],
  "query": {
    "bool": {
      "must": [
        {
          "range": {
            "create_datetime": {
              "gte": "2022-04-20",
              "lte": "2022-06-20"
            }
          }
        }
      ]
    }
  },
  "sort": [
    {
      "create_datetime": {
        "order": "desc"
      }
    }
  ]
}

这个方案没有任何分布式聚合的精度问题，ES仅返回30条文档的单个字段，数据传输量极小，是优先推荐的实现方式。

方案2：ES侧直接返回聚合结果（适合无法调整业务逻辑的场景） #

如果必须单条DSL直接返回sum结果，用scripted_metric聚合实现分片预排序+全局归并求和，逻辑是每个分片先取本地排序后的前30条，汇总到协调节点后再做全局排序取最终前30条计算总和，DSL如下：

{
  "size": 0,
  "query": {
    "bool": {
      "must": [
        {
          "range": {
            "create_datetime": {
              "gte": "2022-04-20",
              "lte": "2022-06-20"
            }
          }
        }
      ]
    }
  },
  "aggs": {
    "sum_top30_duration": {
      "scripted_metric": {
        "init_script": "state.docs = new ArrayList();",
        "map_script": """
          state.docs.add(new HashMap(){{
            put('dt', doc['create_datetime'].value.toInstant().toEpochMilli());
            put('dur', doc['duration_seconds'].value);
          }});
          // 分片内按时间降序排序，仅保留前30条减少跨节点传输量
          state.docs.sort((a,b) -> b.dt.compareTo(a.dt));
          if(state.docs.size() > 30){
            state.docs.remove(30);
          }
        """,
        "combine_script": "return state.docs;",
        "reduce_script": """
          List allDocs = new ArrayList();
          for(shardDocs in states){
            allDocs.addAll(shardDocs);
          }
          // 全局排序后取最新30条求和
          allDocs.sort((a,b) -> b.dt.compareTo(a.dt));
          long total = 0;
          int calCount = Math.min(30, allDocs.size());
          for(int i=0; i<calCount; i++){
            total += allDocs.get(i).dur;
          }
          return total;
        """
      }
    }
  }
}

注意使用这个方案要提前打开脚本支持，保证create_datetime是date类型、duration_seconds是数值类型。

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