1. 背景

此处将单记录一下 global 、 filters和cardinality的聚合操作。

2、解释

1、global

global聚合是全局聚合，是对所有的文档进行聚合，而不受查询条件的限制。

global 聚合器只能作为顶级聚合器，因为将一个 global 聚合器嵌入另一个桶聚合器是没有意义的。

比如： 我们有50个文档，通过查询条件筛选之后存在10个文档，此时我想统计总共有多少个文档。是50个，因为global统计不受查询条件的限制。

2、filters

定义一个多桶聚合，其中每个桶都与一个过滤器相关联。每个桶都会收集与其关联的过滤器匹配的所有文档。

比如： 我们总共有50个文档，通过查询条件筛选之后存在10个文档，此时我想统计 这10个文档中，出现info词语的文档有多少个，出现warn词语的文档有多少个。

3、cardinality

类似于 SQL中的 COUNT(DISTINCT(字段))，不过这个是近似统计，是基于 HyperLogLog++ 来实现的。

3、需求

我们有一组日志，每条日志都存在id和message2个字段。此时根据message字段过滤出存在info warn的日志，然后进行统计：

系统中总共有多少条日志(global + cardinality)

info和warn级别的日志各有多少条(filters)

4、前置条件

4.1 创建mapping

PUT /index_api_log { "settings": { "number_of_shards": 1 }, "mappings": { "properties": { "message":{ "type": "text" }, "id": { "type": "long" } } } }

4.2 准备数据

PUT /index_api_log/_bulk {"index":{"_id":1}} {"message": "this is info message-01","id":1} {"index":{"_id":2}} {"message": "this is info message-02","id":2} {"index":{"_id":3}} {"message": "this is warn message-01","id":3} {"index":{"_id":4}} {"message": "this is error message","id":4} {"index":{"_id":5}} {"message": "this is info and warn message","id":5}

5、实现3的需求

5.1 dsl

POST /index_api_log/_search { "size": 0, "query": { "bool": { "must": [ { "match": { "message": "info warn" } } ] } }, "aggregations": { "agg_01": { "filters": { "filters": { "info": { "match": { "message": "info" } }, "warn": { "match": { "message": "warn" } } }, "other_bucket": true, "other_bucket_key": "other" } }, "agg_02":{ "global": {}, "aggs": { "total": { "cardinality": { "field": "id", "precision_threshold": 30000 } } } } } }

5.2 java 代码

@Test @DisplayName("global and filters and cardinality 聚合") public void test01() throws IOException { SearchRequest request = SearchRequest.of(searchRequest -> searchRequest.index("index_api_log") // 查询 message 中存在 info 和 warn 的日志 .query(query -> query.bool(bool -> bool.must(must -> must.match(match -> match.field("message").query("info warn"))))) // 查询的结果不返回 .size(0) // 第一个聚合 .aggregations("agg_01", agg -> agg.filters(filters -> filters.filters(f -> f.array( Arrays.asList( // 在上一步query的结果中，将 message中包含info的进行聚合 Query.of(q -> q.match(m -> m.field("message").query("info"))), // 在上一步query的结果中，将 message中包含warn的进行聚合 Query.of(q -> q.match(m -> m.field("message").query("warn"))) ) ) ) // 如果上一步的查询中，存在非 info 和 warn的则是否聚合到 other 桶中 .otherBucket(true) // 给 other 桶取一个名字 .otherBucketKey("other") ) ) // 第二个聚合 .aggregations("agg_02", agg -> agg // 此处的 global 聚合只能放在顶部 .global(global -> global) // 子聚合，数据来源于所有的文档，不受上一步query结果的限制 .aggregations("total", subAgg -> // 类似于SQL中的 count(distinct(字段))，是一个近似统计 subAgg.cardinality(cardinality -> // 统计的字段 cardinality.field("id") // 精度，默认值是30000，最大值也是40000，不超过这个值的聚合近似准确值 .precisionThreshold(30000) ) ) ) ); System.out.println("request: " + request); SearchResponse<String> response = client.search(request, String.class); System.out.println("response: " + response); }

5.3 运行结果

6、实现代码

https://gitee.com/huan1993/spring-cloud-parent/blob/master/es/es8-api/src/main/java/com/huan/es8/aggregations/bucket/GlobalAndFiltersAggs.java

7、参考文档

1、https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-bucket-global-aggregation.html