如何在Elasticsearch中按特定严重性统计唯一资产

数据示例 #

[
  {
    "_index": "index",
    "_id": "...",
    "_score": 4.512486,
    "_source": {
      "division": {
        "name": "Ben",
        "id": "678"
      },
      "summary": {
        "severity": 2,
        "last_issue_at": "...",
        "first_issue_at": "...",
        "issues_count": 1
      },
      "company": {
        "id": "..."
      },
      "ingest_timestamp": "..",
      "asset": {
        "name": "Bus",
        "vin": "123",
        "id": "...",
        "manufacturer": "..."
      },
      "timestamp": "..."
    }
  },
  {
    "_index": "index",
    "_id": "...",
    "_score": 4.512486,
    "_source": {
      "division": {
        "name": "Ben",
        "id": "678"
      },
      "summary": {
        "severity": 1,
        "last_issue_at": "...",
        "first_issue_at": "...",
        "issues_count": 1
      },
      "company": {
        "id": "..."
      },
      "ingest_timestamp": "..",
      "asset": {
        "name": "Bus",
        "vin": "123",
        "id": "...",
        "manufacturer": "..."
      },
      "timestamp": "..."
    }
  }
]

统计目标 #

统计严重性为1或2的唯一资产，若某资产同时存在两种严重性，仅按严重性1统计。

当前查询 #

{
  "query": {
    "bool": {
      "must": [
        {
          "match": {
            "company.id": "..."
          }
        },
        {
          "range": {
            "timestamp": {
              "gte": "...",
              "lte": "..."
            }
          }
        }
      ]
    }
  },
  "runtime_mappings": {
    "asset_id_and_vin": {
      "type": "keyword",
      "script": {
        "source": "if (doc.containsKey('asset.vin')) { emit(doc['asset.id'] + ' ' + doc['asset.vin']) } else { emit(doc['asset.id'].value + ' ' + 'N/A') }"
      }
    }
  },
  "aggs": {
    "asset_count": {
      "cardinality": {
        "field": "asset_id_and_vin"
      }
    },
    "assets": {
      "multi_terms": {
        "terms": [
          {
            "field": "asset.name"
          },
          {
            "field": "asset.vin",
            "missing": "N/A"
          },
          {
            "field": "division.name"
          },
          {
            "field": "asset.id"
          },
          {
            "field": "division.id"
          }
        ],
        "order": {
          "earliest_date": "desc"
        },
        "size": 100
      },
      "aggs": {
        "severity": {
          "min": {
            "field": "summary.severity"
          }
        },
        "earliest_date": {
          "min": {
            "field": "summary.first_issue_at"
          }
        },
        "latest_date": {
          "max": {
            "field": "summary.last_issue_at"
          }
        },
        "issues": {
          "sum": {
            "field": "summary.issues_count"
          }
        },
        "filtered_assets": {
          "bucket_selector": {
            "buckets_path": {
              "min_severity": "severity"
            },
            "script": "params.min_severity == 1 || params.min_severity == 2"
          }
        },
        "pagination": {
          "bucket_sort": {
            "size": 100,
            "from": 0
          }
        }
      }
    }
  }
}

问题描述 #

当前查询通过bucket_selector过滤严重性为1或2的结果，但无法正确处理同时存在两种严重性的资产——因为bucket_selector是逐文档检查逻辑，而我们需要对同一资产，只要存在严重性1的记录，就按1统计且只保留该资产的一条统计记录。

优化方案 #

核心思路是对的：通过min聚合取资产的最低严重性（1 < 2，只要资产有严重性1的记录，min结果就是1）。以下是具体调整方案：

1. 顶层查询提前过滤无效数据 #

在query的must中直接添加严重性过滤，减少后续聚合的数据量，提升性能：

{
  "bool": {
    "should": [
      {"term": {"summary.severity": 1}},
      {"term": {"summary.severity": 2}}
    ],
    "minimum_should_match": 1
  }
}

2. 修正基数统计逻辑 #

原来的asset_count会统计所有符合时间和公司条件的资产，现在要确保只统计严重性在1/2范围内的唯一资产，不过因为已经在查询阶段过滤了无效数据，直接保留cardinality即可。

3. 完整优化后的查询 #

{
  "query": {
    "bool": {
      "must": [
        {"match": {"company.id": "..."}},
        {"range": {"timestamp": {"gte": "...", "lte": "..."}}},
        {
          "bool": {
            "should": [
              {"term": {"summary.severity": 1}},
              {"term": {"summary.severity": 2}}
            ],
            "minimum_should_match": 1
          }
        }
      ]
    }
  },
  "runtime_mappings": {
    "asset_id_and_vin": {
      "type": "keyword",
      "script": {
        "source": "if (doc.containsKey('asset.vin')) { emit(doc['asset.id'] + ' ' + doc['asset.vin']) } else { emit(doc['asset.id'].value + ' ' + 'N/A') }"
      }
    }
  },
  "aggs": {
    "asset_count": {
      "cardinality": {
        "field": "asset_id_and_vin"
      }
    },
    "assets": {
      "multi_terms": {
        "terms": [
          {"field": "asset.name"},
          {"field": "asset.vin", "missing": "N/A"},
          {"field": "division.name"},
          {"field": "asset.id"},
          {"field": "division.id"}
        ],
        "order": {"earliest_date": "desc"},
        "size": 100
      },
      "aggs": {
        "min_severity": {
          "min": {"field": "summary.severity"}
        },
        "earliest_date": {
          "min": {"field": "summary.first_issue_at"}
        },
        "latest_date": {
          "max": {"field": "summary.last_issue_at"}
        },
        "total_issues": {
          "sum": {"field": "summary.issues_count"}
        },
        "keep_valid_assets": {
          "bucket_selector": {
            "buckets_path": {"severity": "min_severity"},
            "script": "params.severity == 1 || params.severity == 2"
          }
        },
        "pagination": {
          "bucket_sort": {"size": 100, "from": 0}
        }
      }
    }
  }
}

关键优化点 #

• 查询阶段提前过滤严重性不在1/2范围内的文档，减少聚合计算量。
• 保留min聚合获取资产的最低严重性，确保同时存在两种严重性的资产最终按1统计。
• bucket_selector过滤逻辑保留，进一步确保只输出符合要求的资产分组。
• 调整聚合名称（如issues改为total_issues），让语义更清晰。

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