Elasticsearch中能否合并连续integer_range文档为最大范围？

Elasticsearch 没有原生查询语句能直接实现连续integer_range的合并，但可以通过**脚本化指标聚合（Scripted Metric Aggregation）**或者应用层处理来完成需求。

方案1：脚本化指标聚合（实时查询场景） #

这个聚合允许自定义逻辑处理文档数据，核心步骤是收集所有范围、排序、合并连续区间：

POST /your_index/_search
{
  "size": 0,
  "aggs": {
    "merged_ranges": {
      "scripted_metric": {
        "init_script": "state.ranges = []",
        "map_script": """
          def range = doc['range'].value;
          state.ranges.add(['gte': range.gte, 'lt': range.lt]);
        """,
        "combine_script": """
          // 按gte从小到大排序
          state.ranges.sort((a, b) -> a.gte.compareTo(b.gte));
          def merged = [];
          if (state.ranges.size() == 0) {
            return merged;
          }
          // 初始化第一个合并范围
          def current = state.ranges[0];
          for (int i = 1; i < state.ranges.size(); i++) {
            def next = state.ranges[i];
            // 判断是否连续：当前范围的lt等于下一个范围的gte
            if (current.lt == next.gte) {
              current.lt = next.lt;
            } else {
              merged.add(current);
              current = next;
            }
          }
          merged.add(current);
          return merged;
        """,
        "reduce_script": """
          def finalMerged = [];
          // 合并多节点的聚合结果（单节点可简化）
          for (def nodeResult : states) {
            finalMerged.addAll(nodeResult);
          }
          // 二次排序合并确保结果正确
          finalMerged.sort((a, b) -> a.gte.compareTo(b.gte));
          def result = [];
          if (finalMerged.size() == 0) {
            return result;
          }
          def current = finalMerged[0];
          for (int i = 1; i < finalMerged.size(); i++) {
            def next = finalMerged[i];
            if (current.lt == next.gte) {
              current.lt = next.lt;
            } else {
              result.add(current);
              current = next;
            }
          }
          result.add(current);
          return result;
        """
      }
    }
  }
}

结果说明 #

查询结果会在aggregations.merged_ranges.value中返回合并后的范围数组，对应你给出的示例数据，结果为：

[{"gte":0,"lt":25},{"gte":30,"lt":60}]

方案2：应用层处理（简单易维护） #

如果数据量不大，直接查询所有包含range字段的文档，在应用代码中完成合并逻辑更高效。以Python为例：

# 假设已通过Elasticsearch客户端获取到所有文档的range列表
ranges = [
    {"gte": 0, "lt": 10},
    {"gte": 10, "lt": 25},
    {"gte": 30, "lt": 40},
    {"gte": 40, "lt": 60}
]

# 按gte排序
ranges.sort(key=lambda x: x["gte"])

merged = []
if ranges:
    current = ranges[0].copy()
    for r in ranges[1:]:
        if current["lt"] == r["gte"]:
            current["lt"] = r["lt"]
        else:
            merged.append(current)
            current = r.copy()
    merged.append(current)

print(merged)
# 输出: [{'gte': 0, 'lt': 25}, {'gte': 30, 'lt': 60}]

注意事项 #

• 脚本化聚合需要开启Elasticsearch的脚本支持（在elasticsearch.yml中设置script.allowed_types: inline, stored）
• 大数据量场景下，脚本化聚合可能存在性能瓶颈，建议在数据写入阶段通过Ingest Pipeline预处理合并，或定期执行批处理任务完成合并

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