如何通过单个Elasticsearch查询实现price字段等宽桶聚合统计？

可以通过Elasticsearch的Scripted Metric Aggregation实现单次查询完成需求，无需分两次查询获取min/max再计算间隔。以下是具体实现方案：

查询示例 #

{
  "size": 0,
  "aggs": {
    "global_price_stats": {
      "global": {},
      "aggs": {
        "equal_width_buckets": {
          "scripted_metric": {
            "init_script": "state.prices = []",
            "map_script": "state.prices.add(doc['price'].value)",
            "combine_script": "return state.prices",
            "reduce_script": """
              def allPrices = [];
              buckets.forEach(bucket -> allPrices.addAll(bucket));
              if (allPrices.isEmpty()) {
                return [];
              }
              def minPrice = Collections.min(allPrices);
              def maxPrice = Collections.max(allPrices);
              def bucketCount = 10; // 可按需修改桶的数量
              def interval = (maxPrice - minPrice) / bucketCount;
              
              // 处理所有文档价格相同的边界情况
              if (interval == 0) {
                return [{"key": "${minPrice}-${maxPrice}", "doc_count": allPrices.size()}];
              }
              
              // 初始化所有桶，计数默认0
              def bucketsMap = new HashMap();
              for (int i = 0; i < bucketCount; i++) {
                def start = minPrice + i * interval;
                def end = start + interval;
                // 确保最后一个桶的上限等于maxPrice
                if (i == bucketCount - 1) {
                  end = maxPrice;
                }
                def key = String.format("%.0f-%.0f", start, end);
                bucketsMap.put(key, 0);
              }
              
              // 统计每个价格对应的桶
              allPrices.forEach(price -> {
                def bucketIndex = (int) Math.floor((price - minPrice) / interval);
                // 处理刚好等于maxPrice的价格，归入最后一个桶
                if (bucketIndex >= bucketCount) {
                  bucketIndex = bucketCount - 1;
                }
                def start = minPrice + bucketIndex * interval;
                def end = start + interval;
                if (bucketIndex == bucketCount - 1) {
                  end = maxPrice;
                }
                def key = String.format("%.0f-%.0f", start, end);
                bucketsMap.put(key, bucketsMap.get(key) + 1);
              });
              
              // 生成有序的桶结果列表
              def result = [];
              for (int i = 0; i < bucketCount; i++) {
                def start = minPrice + i * interval;
                def end = start + interval;
                if (i == bucketCount - 1) {
                  end = maxPrice;
                }
                def key = String.format("%.0f-%.0f", start, end);
                result.add({"key": key, "doc_count": bucketsMap.get(key)});
              }
              return result;
            """
          }
        }
      }
    }
  }
}

方案说明 #

1. size: 0：仅返回聚合结果，不返回原始文档，节省资源。
2. global聚合：确保统计所有符合条件的文档（如需过滤，可在global外部添加query节点）。
3. Scripted Metric 阶段解析：
   • init_script：在每个分片上初始化数组，用于存储该分片的所有price值。
   • map_script：遍历分片内的每个文档，将price值加入数组。
   • combine_script：合并分片内的price数组，传递给reduce阶段。
   • reduce_script：合并所有分片的数组，计算price的min/max，自动生成等宽桶，统计每个桶的文档数，空桶默认填充0，最后返回有序的桶列表。

注意事项 #

• 若文档量极大，该方案会在reduce阶段加载所有price值到内存，可能引发内存压力。此时更推荐分两次查询：先通过stats聚合获取min/max，再用histogram聚合（设置min_doc_count: 0和extended_bounds）完成统计，性能更优。
• 可通过修改bucketCount变量调整桶的数量，适配不同需求。
• 脚本已处理边界情况：当所有文档价格相同时，直接生成一个包含全部文档的桶。

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