如何用elasticsearch-dsl定义Runtime Field并按组织聚合求最大扫描时长？

方案1：通过search.extra()注入Runtime Field定义 #

elasticsearch-dsl虽未提供直接定义Runtime Field的封装API，但可通过extra()方法直接添加原生Elasticsearch配置，实现Runtime Field的定义，之后即可对该字段执行聚合操作。

示例代码：

from elasticsearch_dsl import Search, A

# 初始化Search对象
s = Search(using=your_es_client, index="your_asset_index")

# 注入Runtime Field配置，计算scan_age（timestamp与scan_date的毫秒差值）
s = s.extra(
    runtime_mappings={
        "scan_age": {
            "type": "long",
            "script": {
                "source": "doc['timestamp'].value.millis - doc['scan_date'].value.millis"
            }
        }
    }
)

# 定义聚合逻辑：按organization分桶，求scan_age的最大值
s.aggs.bucket(
    "group_by_organization",
    "terms",
    field="organization"
).metric(
    "oldest_scan_age",
    "max",
    field="scan_age"
)

# 执行查询并解析结果
response = s.execute()
for org_bucket in response.aggregations.group_by_organization.buckets:
    print(f"组织: {org_bucket.key}, 最长扫描时长(毫秒): {org_bucket.oldest_scan_age.value}")

方案2：在聚合中直接使用脚本计算（无需单独定义Runtime Field） #

如果仅需在单个聚合中使用差值计算，可直接在max度量中嵌入脚本，省去Runtime Field的定义步骤，更轻量化：

from elasticsearch_dsl import Search, A

s = Search(using=your_es_client, index="your_asset_index")

# 直接在聚合脚本中计算时间差值并求最大值
s.aggs.bucket(
    "group_by_organization",
    "terms",
    field="organization"
).metric(
    "oldest_scan_age",
    "max",
    script={
        "source": "doc['timestamp'].value.millis - doc['scan_date'].value.millis"
    }
)

# 执行查询并解析结果
response = s.execute()
for org_bucket in response.aggregations.group_by_organization.buckets:
    print(f"组织: {org_bucket.key}, 最长扫描时长(毫秒): {org_bucket.oldest_scan_age.value}")

补充说明 #

• 方案1的优势在于scan_age字段可复用，若后续需要对该字段执行过滤、多聚合操作时，只需定义一次即可；
• 脚本中使用millis是因为Elasticsearch日期字段默认存储为毫秒级时间戳，若需转换为秒/分钟等单位，可在脚本中添加除法运算（如/ 1000转为秒）；
• 确保timestamp和scan_date字段在Elasticsearch中为date类型，否则脚本会执行失败。

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