如何将Elasticsearch聚合查询结果存入ES并提取Buckets键值

一、通过Logstash实现（适合定时/持续处理） #

1. 配置ES聚合查询输入 #

修改你的pipeline.conf，添加elasticsearch输入插件直接执行聚合查询：

input {
  elasticsearch {
    hosts => ["http://your-es-address:9200"]
    index => "your-source-index" # 聚合查询的源索引
    query => '{
      "size": 0, # 仅返回聚合结果，跳过原始文档
      "aggs": {
        "custom_agg": { # 替换为你实际的聚合名称
          "terms": {
            "field": "target_field.keyword"
          }
        }
      }
    }'
    schedule => "* * * * *" # 按需设置执行频率，示例为每分钟一次
    codec => "json"
  }
}

2. 过滤提取并拆分buckets #

添加filter段，提取聚合结果中的buckets数组，拆分为单个事件并展开字段：

filter {
  # 提取聚合结果里的buckets数组
  mutate {
    add_field => { "raw_buckets" => "%{[aggregations][custom_agg][buckets]}" }
    remove_field => ["@timestamp", "@version", "aggregations"] # 清理无关字段
  }
  # 将buckets数组拆分为独立事件
  split {
    field => "raw_buckets"
  }
  # 展开每个桶的具体字段（如key、doc_count）
  mutate {
    add_field => {
      "bucket_key" => "%{[raw_buckets][key]}"
      "document_count" => "%{[raw_buckets][doc_count]}"
    }
    remove_field => ["raw_buckets"] # 移除临时字段
  }
}

3. 输出到目标ES索引 #

添加output段，将处理后的结果写入指定索引：

output {
  elasticsearch {
    hosts => ["http://your-es-address:9200"]
    index => "aggregated-results-index" # 存储聚合结果的目标索引
    document_id => "%{bucket_key}" # 可选，用桶key作为文档ID避免重复数据
  }
  stdout { codec => rubydebug } # 调试用，上线后可删除
}

二、通过curl+Shell脚本实现（适合临时一次性处理） #

如果只需要单次执行，用curl配合jq工具处理：

# 执行聚合查询并提取buckets数组到本地文件
curl -X GET "http://your-es-address:9200/your-source-index/_search?size=0" \
  -H "Content-Type: application/json" \
  -d '{
    "aggs": {
      "custom_agg": {
        "terms": {
          "field": "target_field.keyword"
        }
      }
    }
  }' | jq '.aggregations.custom_agg.buckets' > buckets.json

# 遍历buckets数组，逐个写入目标ES索引
jq -c '.[]' buckets.json | while read bucket_doc; do
  curl -X POST "http://your-es-address:9200/aggregated-results-index/_doc/" \
    -H "Content-Type: application/json" \
    -d "$bucket_doc"
done

注意：需要提前安装jq工具用于JSON解析。

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