如何在Elasticsearch中按START/STOP事件聚合日志消息到对应桶？

由于Elasticsearch原生聚合无法直接关联前后文档的START/STOP事件，核心思路是先给每个日志文档打上所属桶的标识ID，再基于该ID做分组聚合。以下是三种可行方案：

方案一：Ingest Pipeline预处理（写入时标记） #

在日志写入ES时，通过Ingest Pipeline给每个文档添加bucket_id字段，实时标记所属桶：

• 遇到标识字段=START时，生成新的桶ID；
• 后续文档（直到标识字段=STOP）沿用当前桶ID；
• 遇到STOP时，标记桶结束，下一个START触发新ID。

示例Painless脚本（Pipeline的处理器部分）：

// 初始化节点级状态
if (ctx._ingest?.bucket_id == null) {
  ctx._ingest.bucket_id = 0;
  ctx._ingest.in_bucket = false;
}

// 处理START事件
if (ctx.标识字段 == "START") {
  ctx._ingest.bucket_id += 1;
  ctx.bucket_id = ctx._ingest.bucket_id;
  ctx._ingest.in_bucket = true;
} 
// 处理STOP事件
else if (ctx.标识字段 == "STOP") {
  ctx.bucket_id = ctx._ingest.bucket_id;
  ctx._ingest.in_bucket = false;
} 
// 中间消息继承当前桶ID
else {
  ctx.bucket_id = ctx._ingest.in_bucket ? ctx._ingest.bucket_id : null;
}

注意：Ingest Pipeline的状态是节点级别的，多节点集群需额外处理状态同步，或改用Transform处理历史数据。

方案二：Transform批量处理历史数据 #

如果日志已写入ES，可用Transform生成带bucket_id的新索引，再基于该索引做聚合：

1. 创建Transform，按时间戳排序日志；
2. 用scripted_metric聚合跟踪START/STOP状态，给每个文档分配桶ID；
3. 输出到新索引后，即可用terms聚合按bucket_id分组。

示例Transform核心配置：

{
  "source": { "index": ["your-log-index"] },
  "dest": { "index": "log-with-buckets" },
  "pivot": {
    "group_by": {
      "timestamp": {
        "date_histogram": {
          "field": "timestamp",
          "fixed_interval": "1ms" // 保证按时间顺序处理每个文档
        }
      }
    },
    "aggregations": {
      "bucketed_docs": {
        "scripted_metric": {
          "init_script": "state.bucket_id = 0; state.in_bucket = false; state.docs = [];",
          "map_script": """
            def doc_data = new HashMap();
            doc_data.putAll(doc);
            if (doc['标识字段'].value == 'START') {
              state.bucket_id += 1;
              state.in_bucket = true;
            } else if (doc['标识字段'].value == 'STOP') {
              state.in_bucket = false;
            }
            doc_data['bucket_id'] = state.in_bucket ? state.bucket_id : null;
            state.docs.add(doc_data);
          """,
          "combine_script": "return state.docs;",
          "reduce_script": """
            def all_docs = [];
            for (batch in states) {
              all_docs.addAll(batch);
            }
            return all_docs;
          """
        }
      }
    }
  }
}

方案三：客户端滚动搜索聚合 #

若不想修改ES索引，可在客户端按时间顺序遍历日志，手动维护桶ID并聚合：

示例Python伪代码：

from elasticsearch import Elasticsearch

es = Elasticsearch()
# 初始化滚动搜索
response = es.search(
    index="your-log-index",
    scroll="1m",
    sort="timestamp:asc",
    query={"match_all": {}}
)

current_bucket_id = 0
in_bucket = False
buckets = {}

# 滚动遍历所有日志
while response["hits"]["hits"]:
    for hit in response["hits"]["hits"]:
        doc = hit["_source"]
        if doc["标识字段"] == "START":
            current_bucket_id += 1
            in_bucket = True
            buckets[current_bucket_id] = []
        elif doc["标识字段"] == "STOP":
            in_bucket = False
        # 将消息加入当前桶
        if in_bucket:
            buckets[current_bucket_id].append(doc)
    # 获取下一批数据
    response = es.scroll(scroll_id=response["_scroll_id"], scroll="1m")

# 输出聚合结果
for bucket_id, messages in buckets.items():
    print(f"Bucket {bucket_id}:")
    for msg in messages:
        print(msg)

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