Elasticsearch 7.16聚合查询如何添加关联database字段？

我正在使用Python 3.6操作Elasticsearch 7.16，Elasticsearch中存储的数据如下：

{"owner": "john", "database": "postgres", "table": "sales_tab"},
{"owner": "hannah", "database": "mongodb", "table": "dept_tab"},
{"owner": "peter", "database": "mysql", "table": "new_tab"},
{"owner": "jim", "database": "postgres", "table": "cust_tab"},
{"owner": "lima", "database": "postgres", "table": "sales_tab"},
{"owner": "tory", "database": "oracle", "table": "store_tab"},
{"owner": "kane", "database": "mysql", "table": "trasit_tab"},
{"owner": "roma", "database": "mongodb", "table": "common_tab"},
{"owner": "ashley", "database": "mongodb", "table": "common_tab"},

执行以下聚合查询：

{
    "size": 0,
    "aggs": {
        "table_grouped": {
          "terms": {
            "field": "table",
            "size": 100000
          }
        }
      }
}

得到的结果仅包含distinct table值及对应文档数：

{... 'aggregations': {'table_grouped': {'doc_count_error_upper_bound': 0, 'sum_other_doc_count': 0, 
'buckets': [{'key': 'sales_tab', 'doc_count': 3}, {'key': 'dept_tab', 'doc_count': 1}, 
{'key': 'new_tab', 'doc_count': 1}, {'key': 'cust_tab', 'doc_count': 1}, 
{'key': 'store_tab', 'doc_count': 1}, {'key': 'trasit_tab', 'doc_count': 1}, 
{'key': 'common_tab', 'doc_count': 2}]}}}

但我需要在每个bucket中添加对应table所属的database字段，期望结果如下：

{... 'aggregations': {'table_grouped': {'doc_count_error_upper_bound': 0, 'sum_other_doc_count': 0, 
'buckets': [{'key': 'sales_tab', 'doc_count': 2, "database": "postgres"}, {'key': 'dept_tab', 
'doc_count': 1, "database": "mongodb"}, {'key': 'new_tab', 'doc_count': 1, 
"database": "mysql"}, {'key': 'cust_tab', 'doc_count': 1, "database": "postgres"}, 
{'key': 'store_tab', 'doc_count': 1, "database": "oracle"}, {'key': 'trasit_tab', 'doc_count': 1, "database": "mysql"}, 
{'key': 'common_tab', 'doc_count': 2, "database": "mongodb"}]}}}

即获取distinct table的同时，得到其对应的database信息，请问该如何实现？

可以通过在terms聚合内部嵌套子聚合来获取对应的database信息，以下两种方案都能满足需求：

方案一：嵌套terms聚合（适合单个table对应唯一database的场景） #

在按table分组的聚合下，添加一个按database分组的子聚合，限制返回数量为1，这样就能拿到每个table对应的database：

{
    "size": 0,
    "aggs": {
        "table_grouped": {
            "terms": {
                "field": "table",
                "size": 100000
            },
            "aggs": {
                "db_info": {
                    "terms": {
                        "field": "database",
                        "size": 1
                    }
                }
            }
        }
    }
}

结果处理 #

执行查询后，每个table的bucket里会包含db_info字段，从中提取第一个bucket的key就是对应的database。在Python中可以这样处理结果：

from elasticsearch import Elasticsearch

es = Elasticsearch(["your-es-host:port"])
response = es.search(index="your-index-name", body=上述查询语句)

# 处理聚合结果
processed_buckets = []
for bucket in response["aggregations"]["table_grouped"]["buckets"]:
    db = bucket["db_info"]["buckets"][0]["key"]
    processed_buckets.append({
        "key": bucket["key"],
        "doc_count": bucket["doc_count"],
        "database": db
    })

# 替换原buckets
response["aggregations"]["table_grouped"]["buckets"] = processed_buckets
print(response)

方案二：使用top_hits聚合（通用场景，支持单个table对应多个database的情况） #

top_hits聚合会返回每个分组下的指定数量文档，这里取第一条文档的database字段即可：

{
    "size": 0,
    "aggs": {
        "table_grouped": {
            "terms": {
                "field": "table",
                "size": 100000
            },
            "aggs": {
                "db_info": {
                    "top_hits": {
                        "size": 1,
                        "_source": ["database"]
                    }
                }
            }
        }
    }
}

结果处理 #

在Python中提取database的方式如下：

processed_buckets = []
for bucket in response["aggregations"]["table_grouped"]["buckets"]:
    db = bucket["db_info"]["hits"]["hits"][0]["_source"]["database"]
    processed_buckets.append({
        "key": bucket["key"],
        "doc_count": bucket["doc_count"],
        "database": db
    })

response["aggregations"]["table_grouped"]["buckets"] = processed_buckets
print(response)

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