Elasticsearch 5.6嵌套数据去重与聚合实现方案咨询

我在Elasticsearch 5.6中有包含嵌套类型malicious_family的数据，示例数据如下：

{
  "_id": "AYc1Xk4JMTFNgTcwpES6",
  "_source": {
    "task_id": "03fe407b9754506a28dbaf71d01c5cf41dd65d479cc7198c19bd7efe",
    "sha1": "fdbd6026f6a4061f3a83f6242f66ae72",
    "detect_status": 1,
    "malicious_family": [
      {
        "score": 1,
        "name": "gameteaspy",
        "analysis_engine": ["RAS"],
        "data_source": ["qax"]
     },
      {
        "score": 1,
        "name": "malware",
        "analysis_engine": ["RAS"],
        "data_source": ["qax"]
      },
      {
        "score": 0.9718849643766059,
        "name": "zbot",
        "analysis_engine": ["ML"],
        "data_source": ["qax"]
      }
    ]
  }
}

需要实现以下需求：

• 对malicious_family.name进行聚合
• 过滤掉malicious_family.score < 1.0的条目
• 按sha1去重，即同一个样本（sha1唯一）下的同一家族只算一次

最终预期结果：

[{'name': 'gameteaspy', 'sample_count': 1}, {'name': 'malware', 'sample_count': 1}]

由于malicious_family是嵌套类型，必须使用**嵌套聚合（Nested Aggregation）**处理内部字段，结合过滤和基数聚合实现去重统计，具体DSL查询如下：

{
  "size": 0, // 无需返回原始文档，仅获取聚合结果
  "aggs": {
    // 进入嵌套字段上下文
    "nested_families": {
      "nested": {
        "path": "malicious_family"
      },
      "aggs": {
        // 过滤出score≥1.0的家族条目
        "score_filter": {
          "filter": {
            "range": {
              "malicious_family.score": {
                "gte": 1.0
              }
            }
          },
          "aggs": {
            // 按家族名称分组
            "family_groups": {
              "terms": {
                "field": "malicious_family.name",
                "size": 10000 // 根据实际数据量调整，避免结果截断
              },
              "aggs": {
                // 统计每个家族对应的唯一样本数（按sha1去重）
                "unique_sample_count": {
                  "cardinality": {
                    "field": "sha1"
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}

结果转换 #

执行查询后，聚合结果会返回在aggregations.nested_families.score_filter.family_groups.buckets中，每个桶包含家族名称和去重后的样本数。用Python转换为预期格式的示例代码：

# 假设response是ES查询返回的结果
buckets = response['aggregations']['nested_families']['score_filter']['family_groups']['buckets']
result = [{'name': bucket['key'], 'sample_count': bucket['unique_sample_count']['value']} for bucket in buckets]
print(result)

关键说明 #

• 嵌套聚合的必要性：嵌套类型字段独立于根文档存储，必须通过nested聚合进入其上下文，才能正确过滤和聚合内部字段。
• 基数聚合（Cardinality）：用于统计sha1的唯一值数量，实现按样本去重的需求。
• size参数调整：terms聚合的size需设置足够大，确保所有家族名称都被返回，避免结果截断。

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