Elasticsearch：按进程名分组统计各进程最新状态数量

解决方案 #

要统计每个进程最新状态的数量，你可以通过以下两种方式实现：

方法一：使用 collapse 简化查询

利用 Elasticsearch 的 collapse 功能按进程名称分组，仅保留每个进程最新时间戳的文档，再对这些文档的状态进行聚合统计：

GET /process_logs*/_search
{
  "size": 0,
  "query": {
    "exists": {
      "field": "process_status"
    }
  },
  "collapse": {
    "field": "process_attributes.process_name.keyword",
    "inner_hits": {
      "size": 1,
      "sort": [{"@timestamp": "desc"}]
    }
  },
  "aggs": {
    "status_counts": {
      "terms": {
        "field": "process_status.keyword",
        "size": 2
      }
    }
  }
}

说明：

• collapse 按 process_name.keyword 分组，inner_hits 配置仅返回每组中时间戳最新的文档。
• 聚合 status_counts 基于这些去重后的最新文档，统计不同状态的进程数量。
• 返回结果的 aggregations.status_counts.buckets 会直接给出 Running 和 Stopped 的数量，与你的 SQL 查询结果一致。

方法二：使用嵌套聚合 + 脚本化指标（适合复杂场景）

如果无法使用 collapse，可以通过嵌套聚合获取每个进程的最新状态，再用脚本化指标聚合统计：

GET /process_logs*/_search
{
  "size": 0,
  "query": {
    "exists": {
      "field": "process_status"
    }
  },
  "aggs": {
    "processes": {
      "terms": {
        "field": "process_attributes.process_name.keyword"
      },
      "aggs": {
        "latest_status": {
          "top_metrics": {
            "metrics": [{"field": "process_status.keyword"}],
            "sort": {"@timestamp": "desc"},
            "size": 1
          }
        }
      }
    },
    "status_summary": {
      "scripted_metric": {
        "init_script": "state.counts = [:]",
        "map_script": "void ctx.processes.buckets.each { bucket -> def status = bucket.latest_status.top[0].process_status.keyword; state.counts[status] = (state.counts[status] ?: 0) + 1 }",
        "combine_script": "return state.counts",
        "reduce_script": "def finalCounts = [:]; states.each { counts -> counts.each { k, v -> finalCounts[k] = (finalCounts[k] ?: 0) + v } }; return finalCounts"
      }
    }
  }
}

说明：

• processes 聚合按进程名称分组，top_metrics 获取每组最新时间戳对应的状态。
• status_summary 脚本化指标遍历所有进程桶，统计各状态的出现次数。
• 返回结果的 aggregations.status_summary.value 会给出各状态的计数。

结果示例 #

两种方法最终都会返回类似如下的统计结果：

"aggregations" : {
  "status_counts" : {
    "buckets" : [
      {
        "key" : "Running",
        "doc_count" : 2
      },
      {
        "key" : "Stopped",
        "doc_count" : 2
      }
    ]
  }
}

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