基于分组最大ID状态计算Elasticsearch状态占比的查询方案

需求说明 #

从给定数据集中，先获取每个workflow对应最大id的status，再计算各状态占唯一workflow总数的百分比。

查询语句（ES 7.10+ 推荐使用） #

使用top_metrics聚合高效获取每个分组的最新状态，再通过脚本计算占比：

{
  "size": 0,
  "aggs": {
    "group_by_workflow": {
      "terms": {
        "field": "workflow.keyword",
        "size": 1000
      },
      "aggs": {
        "latest_status": {
          "top_metrics": {
            "metrics": {"field": "status"},
            "sort": {"id": "desc"}
          }
        }
      }
    },
    "total_workflows": {
      "cardinality": {
        "field": "workflow.keyword"
      }
    },
    "status_percentages": {
      "bucket_script": {
        "buckets_path": {
          "total": "total_workflows.value",
          "statuses": "group_by_workflow>latest_status.status"
        },
        "script": """
          def counts = [:];
          params.statuses.each { status ->
            counts[status] = (counts[status] ?: 0) + 1;
          }
          def percentages = [:];
          counts.each { k, v ->
            percentages[k] = (v / params.total) * 100;
          }
          return percentages;
        """
      }
    }
  }
}

兼容低版本ES（低于7.10）的查询语句 #

用top_hits替代top_metrics：

{
  "size": 0,
  "aggs": {
    "group_by_workflow": {
      "terms": {
        "field": "workflow.keyword",
        "size": 1000
      },
      "aggs": {
        "latest_hit": {
          "top_hits": {
            "size": 1,
            "sort": [{"id": {"order": "desc"}}],
            "_source": ["status"]
          }
        }
      }
    },
    "total_workflows": {
      "cardinality": {
        "field": "workflow.keyword"
      }
    },
    "status_percentages": {
      "bucket_script": {
        "buckets_path": {
          "total": "total_workflows.value",
          "hits": "group_by_workflow>latest_hit.hits.hits"
        },
        "script": """
          def counts = [:];
          params.hits.each { hitList ->
            def status = hitList[0]._source.status;
            counts[status] = (counts[status] ?: 0) + 1;
          }
          def percentages = [:];
          counts.each { k, v ->
            percentages[k] = (v / params.total) * 100;
          }
          return percentages;
        """
      }
    }
  }
}

关键部分解释 #

• size: 0：不返回原始文档，只生成聚合结果
• group_by_workflow：按workflow.keyword精确分组，确保每个workflow单独统计
• latest_status/latest_hit：分别用top_metrics/top_hits获取每个分组中id最大的文档对应的status
• total_workflows：用cardinality聚合统计唯一workflow的总数量
• status_percentages：通过bucket_script脚本统计各状态的出现次数，再除以总workflow数计算百分比

预期结果 #

查询返回的status_percentages字段会包含各状态的占比：

"status_percentages": {
  "value": {
    "SUCCESS": 25.0,
    "FAILURE": 75.0
  }
}

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