ElasticSearch中同状态最新记录的聚合计算求助

我的ElasticSearch索引some_index中有如下数据：

[
  {
    "_index": "some_index",
    "_source": {
      "cart": {
        "cart_id": 1,
        "cart_status": "new",
        "grandTotal": 12,
        "event": "some_event",
        "timestamp": "2022-12-01T00:00:00.000Z"
      }
    }
  },
  {
    "_index": "some_index",
    "_source": {
      "cart": {
        "cart_id": 1,
        "cart_status": "paid",
        "grandTotal": 12,
        "event": "some_event",
        "timestamp": "2022-12-02T00:00:00.000Z"
      }
    }
  },
  {
    "_index": "some_index",
    "_source": {
      "cart": {
        "cart_id": 2,
        "cart_status": "new",
        "grandTotal": 23,
        "event": "some_event",
        "timestamp": "2022-12-01T00:00:00.000Z"
      }
    }
  },
  {
    "_index": "some_index",
    "_source": {
      "cart": {
        "cart_id": 2,
        "cart_status": "paid",
        "grandTotal": 23,
        "event": "some_event",
        "timestamp": "2022-12-04T00:00:00.000Z"
      }
    }
  },
  {
    "_index": "some_index",
    "_source": {
      "cart": {
        "cart_id": 3,
        "cart_status": "new",
        "grandTotal": 17,
        "event": "some_event",
        "timestamp": "2022-12-01T00:00:00.000Z"
      }
    }
  },
  {
    "_index": "some_index",
    "_source": {
      "cart": {
        "cart_id": 3,
        "cart_status": "new",
        "grandTotal": 17,
        "event": "some_event",
        "timestamp": "2022-12-04T00:00:00.000Z"
      }
    }
  }
]

需要实现：在指定时间范围内，按每个cart_id的最新cart_status分组，统计grandTotal的总和。以时间范围timestamp >= 2022-12-01 00:00:00 且 timestamp<= 2022-12-03 00:00:00为例，预期结果是：

• cart_status:new，grandTotal总和为40（cart_id 2和3的最新状态为new）
• cart_status:paid，grandTotal总和为12（仅cart_id 1的最新状态为paid）

尝试过两种方法但都失败：

• 在top_hits聚合上添加子聚合，报错："Aggregator [top_result] of type [top_hits] cannot accept sub-aggregations"
• 使用collapse获取最新状态，但无法对collapse结果进行聚合

要求支持任意时间范围的动态查询，且避免将中间数据存储到其他索引。

可以通过**terms聚合+top_hits+scripted_metric聚合**的组合实现需求，具体DSL如下：

POST some_index/_search
{
  "size": 0,
  "query": {
    "range": {
      "cart.timestamp": {
        "gte": "2022-12-01T00:00:00.000Z",
        "lte": "2022-12-03T00:00:00.000Z"
      }
    }
  },
  "aggs": {
    "group_by_cart": {
      "terms": {
        "field": "cart.cart_id"
      },
      "aggs": {
        "latest_cart": {
          "top_hits": {
            "size": 1,
            "sort": [{"cart.timestamp": "desc"}]
          }
        }
      }
    },
    "sum_by_status": {
      "scripted_metric": {
        "init_script": "state.sums = [:]",
        "map_script": """
          def cartBucket = params._agg.group_by_cart.buckets;
          for (bucket in cartBucket) {
            def latestDoc = bucket.latest_cart.hits.hits[0]._source.cart;
            def status = latestDoc.cart_status;
            def total = latestDoc.grandTotal;
            if (state.sums.containsKey(status)) {
              state.sums[status] += total;
            } else {
              state.sums[status] = total;
            }
          }
        """,
        "combine_script": "return state.sums",
        "reduce_script": """
          def finalSums = [:];
          for (sums in states) {
            for (entry in sums.entrySet()) {
              def status = entry.getKey();
              def total = entry.getValue();
              if (finalSums.containsKey(status)) {
                finalSums[status] += total;
              } else {
                finalSums[status] = total;
              }
            }
          }
          return finalSums;
        """
      }
    }
  }
}

逻辑解释 #

1. 时间范围过滤：通过range查询限定处理的文档范围，只纳入指定时间内的记录。
2. 按cart_id取最新状态：
   • terms聚合按cart_id分组，确保每个购物车只被处理一次。
   • 子聚合top_hits按timestamp倒序排序，取每组第一条文档，即该购物车在时间范围内的最新状态记录。
3. 按状态统计总和：
   • scripted_metric聚合在map阶段遍历所有购物车分组，提取最新记录的状态和金额并累加。
   • combine和reduce阶段合并各分片的计算结果，最终得到各状态对应的grandTotal总和。

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