如何提取Elasticsearch子聚合查询结果生成类SQL分组表格

Elasticsearch 原生能力说明 #

Elasticsearch 没有内置方法可以直接将多层嵌套terms聚合的结果自动转换为扁平二维表结构返回，也不支持将聚合结果自动填充到主结果的_source字段中输出。所有聚合结果默认会严格按照你定义的聚合层级，返回嵌套的bucket结构。

最优实现方案 #

方案1：替换为composite聚合（最推荐） #

你要实现的是类似SQL group by field1,field2,field3 后输出多维度+聚合指标的效果，完全不需要写三层嵌套terms聚合，直接用ES官方为多字段分组场景设计的composite聚合即可，天生返回扁平结构，性能远高于嵌套terms，还支持游标分页拉取全量数据，不会出现你当前写法中每层设置size:10000容易导致的ES内存溢出问题。
查询语句示例：

{
  "query": SEARCH_BODY,
  "size": 0,
  "aggs": {
    "group_result": {
      "composite": {
        "size": 1000,
        "sources": [
          {"aggregate1": {"terms": {"field": "field1"}}},
          {"aggregate2": {"terms": {"field": "field2"}}},
          {"aggregate3": {"terms": {"field": "field3"}}}
        ]
      },
      "aggs": {
        "column1": {
          "sum": {"field": "field4"}
        }
      }
    }
  }
}

该查询返回的聚合结果本身就是扁平结构，不需要多层递归解析，每个bucket直接携带所有分组维度的key和指标值，结构示例：

{
  "aggregations": {
    "group_result": {
      "after_key": {"aggregate1": "A", "aggregate2": "B", "aggregate3": "C"},
      "buckets": [
        {
          "key": {
            "aggregate1": "A",
            "aggregate2": "B",
            "aggregate3": "C"
          },
          "doc_count": 1,
          "column1": {"value": 12345}
        }
      ]
    }
  }
}

拿到结果后只需要一次遍历，就能把每个bucket的key和指标值拼成二维表行数据，直接转成DataFrame或者扁平JSON即可，处理成本极低。如果结果总量超过单次设置的size，只需要把上一次返回的after_key放到下一次请求的composite参数中，就能分页拉取完全量结果。

方案2：现有嵌套结构的轻量解析 #

如果你因为版本兼容等原因必须保留三层嵌套terms的写法，也不需要硬编码每层的解析逻辑，写一个通用的递归遍历方法即可自动提取所有行数据，示例Python实现：

import pandas as pd

def parse_nested_aggs(buckets, dim_names, metric_name, current_row=None):
    rows = []
    current_row = current_row or []
    for bucket in buckets:
        row = current_row.copy()
        row.append(bucket["key"])
        # 命中最内层指标，生成完整行
        if metric_name in bucket:
            row.append(bucket[metric_name]["value"])
            rows.append(row)
            continue
        # 遍历查找下一层聚合的bucket
        for v in bucket.values():
            if isinstance(v, dict) and "buckets" in v:
                rows.extend(parse_nested_aggs(v["buckets"], dim_names, metric_name, row))
                break
    return rows

# 调用方法直接生成DataFrame
raw_buckets = temp["buckets"] # temp是你拿到的最外层聚合结果
df = pd.DataFrame(
    parse_nested_aggs(raw_buckets, ["aggregate1","aggregate2","aggregate3"], "column"),
    columns=["aggregate1","aggregate2","aggregate3","column1"]
)

这个方法是通用的，不管嵌套多少层terms聚合都能正常解析，不需要针对每一层写单独的取值逻辑，代码量很小。

注意事项 #

你当前的查询写法中三层terms都设置size:10000风险极高，三层维度的笛卡尔积最坏情况下会生成1万亿个bucket，极易直接打垮ES集群，除非你能百分百确定三个字段的基数乘积极小，否则强烈建议替换为composite聚合实现。

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