Elasticsearch Bucket聚合(桶聚合) 第一篇(常用桶聚合一览)-阿里云开发者社区

本篇将开始介绍Elasticsearch Bucket聚合(桶聚合)。

Buket Aggregations(桶聚合)不像metrics Aggregations(度量聚合)那样计算字段上的度量，而是创建文档桶，每个文件桶有效地定义一个文档集。除了bucket本身之外，bucket聚合还计算并返回“落入”每个bucket的文档的数量。

与度量聚合相反，桶聚合可以嵌套子聚合。这些子聚合将为它们的“父”桶聚合创建的桶进行聚合。

ES Bucket Aggregations对标关系型数据库的(group by)。

首先我们来介绍桶聚合两个常用参数intervals、time_zone的含义。

1、Intervals

定义桶的间隔，其可选值如下：

seconds
1, 5, 10, 30的倍数。
minutes
1, 5, 10, 30的倍数。
hours
1, 3, 12的倍数。
days
1,7的倍数。
months
1, 3的倍数。
years
1, 5, 10, 20, 50, 100的倍数。

2、Time Zone

对于日期类型，可以使用time_zone来指定时区，可选值可以是相对ISO 8601 utc的相对值，例如+01:00或-08:00，也可以是时区ID，例如America/Los_Angeles。

3、Histogram Aggregation

直方图聚合，Date Histogram Aggregation是其特例。

动态将文档中的值按照特定的间隔构建桶，并计算落在该桶的数量，文档中的值根据如下函数进行近似匹配：

bucket_key = Math.floor((value - offset) / interval) * interval + offset，
其中interval必须是正小数（包含正整数），offset为[0,interval)。

主要支持的参数如下：

keyed
响应结果返回组织方式（数组或对象），具体示例请参考日期类直方图聚合。
doc_count
匹配的文档数量。
offset 偏移量
更改每个bucket(桶)的开始时间，例如将offset设置为"10"，则上例中返回的一个桶的key为：[10,30)，如果offset设置为5，则第一个桶的key为[15,30)。
order
默认按照key的升序进行排序，可以通过order字段来指定排序，其值为BucketOrder。

其取值：

BucketOrder.count(boolean asc)
按匹配文档格式升序/降序排序。
BucketOrder.key(boolean asc)
按key的升序或降序排序。
BucketOrder.aggregation
通过定义一个子聚合进行排序。
BucketOrder.compound(List< BucketOrder> orders)
创建一个桶排序策略，该策略根据多个条件对桶进行排序。

min_doc_count
表示只显示匹配的文档大于等于min_doc_count的桶。

具体JAVA的示例将在Date Histogram Aggregation中详细介绍。

4、Date Histogram Aggregation

日期字段直方图聚合。

4.1 interval 取值

milliseconds (ms)
毫秒，固定长度，支持倍数，通常使用1000的倍数。
seconds (s)
秒
minutes (m)
分钟。所有的分钟从00秒开始

1m，表示在指定时区的第一分钟00s到下一分钟00s之间的时间段。
{n}m，表示时间间隔，等于n 60 1000 毫秒。

hours (h)
小时，其分钟与秒都从00开始。

1小时(1h)是指定时区内第一个小时的00:00分钟到下一个小时的00:00分钟之间的时间间隔，用来补偿其间的任何闰秒，从而使经过该小时的分钟数和秒数在开始和结束时相同。
{n}h，表示时间间隔，等于 n 60 60 * 1000 毫秒的时间间隔。

days (d)

一天(1d)是在指定的时区内，从一天的开始到第二天的开始的时间间隔。
{n}d，表示时间间隔，等于n 24 60 60 1000毫秒。

weeks (w)

1周(1w)为开始日:of_week:hour:minute:second与一周的同一天及下一周的时间在指定时区的间隔。
不支持 {n}w。

months (M)

一个月(1M)是本月开始之间的时间间隔的一天与次月的同一天。
不支持{n}M

quarters (q)
季度，不支持{n}q。
years (y)
年，不支持{n}y。

4.2 示例

{
    "aggs" : {
        "sales_over_time" : {
            "date_histogram" : {
                "field" : "date",
                "interval" : "month"
            }
        }
    }
}

对应的JAVA示例如下：

/**
     * 日期直方图聚合
     */
    public static void test_Date_Histogram_Aggregation() {
        RestHighLevelClient client = EsClient.getClient();
        try {
            
            //构建日期直方图聚合  时间间隔，示例中按月统计
            DateHistogramInterval interval = new DateHistogramInterval("1M"); 
            SearchRequest searchRequest = new SearchRequest();
            searchRequest.indices("aggregations_index02");
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            AggregationBuilder aggregationBuild = AggregationBuilders.dateHistogram("createTime_histogram")
                                                                        .field("createTime")
                                                                        .dateHistogramInterval(interval)
                                                                    //    .format("yyyy-MM-dd") // 对key的格式化
                                                  ;
            sourceBuilder.aggregation(aggregationBuild);
            sourceBuilder.size(0);
            sourceBuilder.query(
                    QueryBuilders.termQuery("sellerId", 24)
            );
            searchRequest.source(sourceBuilder);
            SearchResponse result = client.search(searchRequest, RequestOptions.DEFAULT);
            System.out.println(result);
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            EsClient.close(client);
        }
    }

对应的返回值：

{
    ... //省略常规响应
    "aggregations":{
        "date_histogram#createTime_histogram":{
            "buckets":[
                    "key_as_string":"2015-12-01 00:00:00",
                    "key":1448928000000,
                    "doc_count":6
                },
                {
                    "key_as_string":"2016-01-01 00:00:00",  
                    "key":1451606400000,
                    "doc_count":4
                }
            ]
        }
    }
}

其相应的参数已在上面详述，在此不重复介绍。

4.3 Date Histogram聚合支持的常用参数

除Histogram Aggregation罗列的参数后，还额外支持如下参数：

timeZone 时区指定。
offset 偏移量
更改每个bucket(桶)的开始时间，例如将offset设置为"1h"，则上例中返回的一个桶的开始时间："2015-12-01 00:00:00"，则更改为"2015-12-01 01:00:00"
format
key格式化，将key使用format格式化后的值设置为key_as_string字段。
keyed
返回结果格式化，默认为false，则buckets返回值为数组，如果keyed=true，则对应的返回结果如下：

"aggregations":{
        "date_histogram#createTime_histogram":{
            "buckets":{
                "2015-12-01 00:00:00":{
                    "key_as_string":"2015-12-01 00:00:00",
                    "key":1448928000000,
                    "doc_count":6
                },
                "2016-01-01 00:00:00":{
                    "key_as_string":"2016-01-01 00:00:00",
                    "key":1451606400000,
                    "doc_count":4
                }
            }
        }
    }
}

5、Date Range Aggregation

日期范围聚合，每个范围定义[from,to)，from,to可支持date mesh格式。
其使用示例如下，其他与 Date Histogram类似。

/**
     * 日期范围聚合
     */
    public static void test_Date_range_Aggregation() {
        RestHighLevelClient client = EsClient.getClient();
        try {
            //构建日期直方图聚合  时间间隔，示例中按月统计
            SearchRequest searchRequest = new SearchRequest();
            searchRequest.indices("aggregations_index02");
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            AggregationBuilder aggregationBuild = AggregationBuilders.dateRange("createTime_date_range")
                                                                        .field("createTime")
                                                                        .format("yyyy-MM-dd")
                                                                        .addRange("quarter_01", "2016-01", "2016-03")
                                                                        .addRange("quarter_02", "2016-03", "2016-06")
                                                                        .addRange("quarter_03", "2016-06", "2016-09")
                                                                        .addRange("quarter_04", "2016-09", "2016-12")
                                                                    
                                                                    //    .format("yyyy-MM-dd") // 对key的格式化
                                                  ;
            sourceBuilder.aggregation(aggregationBuild);
            sourceBuilder.size(0);
            sourceBuilder.query(
                    QueryBuilders.termQuery("sellerId", 24)
            );
            searchRequest.source(sourceBuilder);
            SearchResponse result = client.search(searchRequest, RequestOptions.DEFAULT);
            System.out.println(result);
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            EsClient.close(client);
        }
    }

6、Filter Aggregation

聚合中支持首先根据过滤上下文对所有文档进行刷选，然后再进行聚合计算，例如：

POST /sales/_search?size=0
{
    "aggs" : {
        "t_shirts" : {
            "filter" : { "term": { "type": "t-shirt" } },
            "aggs" : {
                "avg_price" : { "avg" : { "field" : "price" } }
            }
        }
    }
}

其对应的JAVA代码如下：

/**
     * 日期范围聚合
     */
    public static void test_filter_Aggregation() {
        RestHighLevelClient client = EsClient.getClient();
        try {
            //构建日期直方图聚合  时间间隔，示例中按月统计
            SearchRequest searchRequest = new SearchRequest();
            searchRequest.indices("aggregations_index02");
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            AggregationBuilder aggregationBuild = AggregationBuilders.filter("t_shirts", QueryBuilders.termQuery("status", "1"))
                                                    .subAggregation(AggregationBuilders.avg("avg").field("num"))
                                                  ;
            sourceBuilder.aggregation(aggregationBuild);
            sourceBuilder.size(0);
            sourceBuilder.query(
                    QueryBuilders.termQuery("sellerId", 24)
            );
            searchRequest.source(sourceBuilder);
            SearchResponse result = client.search(searchRequest, RequestOptions.DEFAULT);
            System.out.println(result);
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            EsClient.close(client);
        }
    }

其返回结果如下：

{
    ... //省略
    "aggregations":{
        "filter#t_shirts":{
            "doc_count":2,
            "avg#avg":{
                "value":1
            }
        }
    }
}

{

... //省略
"aggregations":{
    "filter#t_shirts":{
        "doc_count":2,
        "avg#avg":{
            "value":1
        }
    }
}

}

7、Filters Aggregation

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

public static void test_filters_aggregation() {
        RestHighLevelClient client = EsClient.getClient();
        try {
            //构建日期直方图聚合  时间间隔，示例中按月统计
            SearchRequest searchRequest = new SearchRequest();
            searchRequest.indices("aggregations_index02");
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            AggregationBuilder aggregationBuild = AggregationBuilders.filters("create_filters", 
                                                        QueryBuilders.termQuery("status", 1),
                                                        QueryBuilders.termQuery("buyerId", 1))
                                                    .subAggregation(AggregationBuilders.avg("avg").field("num"))
                                                  ;
            sourceBuilder.aggregation(aggregationBuild);
            sourceBuilder.size(0);
            sourceBuilder.query(
                    QueryBuilders.termQuery("sellerId", 24)
            );
            searchRequest.source(sourceBuilder);
            SearchResponse result = client.search(searchRequest, RequestOptions.DEFAULT);
            System.out.println(result);
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            EsClient.close(client);
        }

    }

其返回结果：

{
    ... // 省略
    "aggregations":{
        "filters#create_filters":{
            "buckets":[
                {
                    "doc_count":2,
                    "avg#avg":{
                        "value":1
                    }
                },
                {
                    "doc_count":0,
                    "avg#avg":{
                        "value":null
                    }
                }
            ]
        }
    }
}

温馨提示，每一个filter代表一个桶（聚合）。

8、Global Aggregation

全局聚合，会忽略所有的查询条件，具体从下述例子进行说明：

POST /sales/_search?size=0
{
    "query" : {
        "match" : { "type" : "t-shirt" }
    },
    "aggs" : {
        "all_products" : {
            "global" : {}, 
            "aggs" : { 
                "avg_price" : { "avg" : { "field" : "price" } }
            }
        },
        "t_shirts": { "avg" : { "field" : "price" } }
    }
}

其聚合的文档集不是匹配该查询的文档"query" : {"match" : { "type" : "t-shirt" } }，而是针对所有的文档进行聚合。

对应的JAVA实例如下：

public static void test_global_aggregation() {
        RestHighLevelClient client = EsClient.getClient();
        try {
            //构建日期直方图聚合  时间间隔，示例中按月统计
            SearchRequest searchRequest = new SearchRequest();
            searchRequest.indices("aggregations_index02");
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            AggregationBuilder aggregationBuild = AggregationBuilders.global("all_producers")
                                                            .subAggregation(AggregationBuilders
                                                                    .avg("num_avg_aggregation")
                                                                    .field("num"))
                                                  ;
            sourceBuilder.aggregation(aggregationBuild);
            sourceBuilder.size(0);
            sourceBuilder.query(
                    QueryBuilders.termQuery("sellerId", 24)
            );
            searchRequest.source(sourceBuilder);
            SearchResponse result = client.search(searchRequest, RequestOptions.DEFAULT);
            System.out.println(result);
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            EsClient.close(client);
        }

    }

对应的返回值如下：

{
    "took":151,
    "timed_out":false,
    "_shards":{
        "total":5,
        "successful":5,
        "skipped":0,
        "failed":0
    },
    "hits":{
        "total":39,                       // @1
        "max_score":0,
        "hits":[

        ]
    },
    "aggregations":{
        "global#all_producers":{
            "doc_count":1286,      // @2
            "avg#num_avg_aggregation":{
                "value":1.3157076205287714
            }
        }
    }
}

结果@1：表示符合查询条件的总个数。
结构@2：表示参与聚合的文档数量，等于当前库中文档总数。

9、IP Range Aggregation

ip类型特有的范围聚合，与其他聚合使用类似，就不重复介绍了。

10、Missing Aggregation

统计缺少某个字段的文档个数。
JAVA示例如下：

AggregationBuilder aggregationBuild = AggregationBuilders.missing("missing_num_count")
                                                        .field("num");

11、Range Aggregation

基于多桶值源的聚合，允许用户定义一组范围——每个范围表示一个桶。在聚合过程中，将根据每个bucket范围和相关/匹配文档的“bucket”检查从每个文档中提取的值。注意，此聚合包含from值，并排除每个范围的to值。

GET /_search
{
    "aggs" : {
        "price_ranges" : {
            "range" : {
                "field" : "price",
                "ranges" : [
                    { "to" : 100.0 },
                    { "from" : 100.0, "to" : 200.0 },
                    { "from" : 200.0 }
                ]
            }
        }
    }
}

对应的JAVA示例如下：

public static void test_range_aggregation() {
        RestHighLevelClient client = EsClient.getClient();
        try {
            //构建日期直方图聚合  时间间隔，示例中按月统计
            SearchRequest searchRequest = new SearchRequest();
            searchRequest.indices("aggregations_index02");
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            AggregationBuilder aggregationBuild = AggregationBuilders.range("num_range_aggregation")
                                                                    .field("num")
                                                                    .addRange(0, 5)
                                                                    .addRange(5,10)
                                                                    .addUnboundedFrom(10)
                                                  ;
            sourceBuilder.aggregation(aggregationBuild);
            sourceBuilder.size(0);
            sourceBuilder.query(
                    QueryBuilders.termQuery("sellerId", 24)
            );
            searchRequest.source(sourceBuilder);
            SearchResponse result = client.search(searchRequest, RequestOptions.DEFAULT);
            System.out.println(result);
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            EsClient.close(client);
        }

    }

其返回结果如下：

{
    // 省略
    "aggregations":{
        "range#num_range_aggregation":{
            "buckets":[
                {
                    "key":"0.0-5.0",
                    "from":0,
                    "to":5,
                    "doc_count":38
                },
                {
                    "key":"5.0-10.0",
                    "from":5,
                    "to":10,
                    "doc_count":0
                },
                {
                    "key":"10.0-*",
                    "from":10,
                    "doc_count":1
                }
            ]
        }
    }
}

Range Aggregations支持嵌套聚合，使用subAggregations来支持嵌套聚合，根据官网示例如下：

GET /_search
{
    "aggs" : {
        "price_ranges" : {
            "range" : {                                    // @1
                "field" : "price",
                "ranges" : [
                    { "to" : 100 },
                    { "from" : 100, "to" : 200 },
                    { "from" : 200 }
                ]
            },
            "aggs" : {                                  // @2
                "price_stats" : {
                    "stats" : { "field" : "price" }
                }
            }
        }
    }
}

首先通过@1定义范围聚合，然后对每个桶中的文档再执行子聚合@2，其返回结果如下：

{
  ...
  "aggregations": {
    "price_ranges": {
      "buckets": [
        {
          "key": "*-100.0",
          "to": 100.0,
          "doc_count": 2,
          "price_stats": {
            "count": 2,
            "min": 10.0,
            "max": 50.0,
            "avg": 30.0,
            "sum": 60.0
          }
        },
        {
          "key": "100.0-200.0",
          "from": 100.0,
          "to": 200.0,
          "doc_count": 2,
          "price_stats": {
            "count": 2,
            "min": 150.0,
            "max": 175.0,
            "avg": 162.5,
            "sum": 325.0
          }
        },
        {
          "key": "200.0-*",
          "from": 200.0,
          "doc_count": 3,
          "price_stats": {
            "count": 3,
            "min": 200.0,
            "max": 200.0,
            "avg": 200.0,
            "sum": 600.0
          }
        }
      ]
    }
  }
}

本文详细介绍了ES 桶聚合，并给出JAVA示例，下一篇将重点关注ES桶聚合之term聚合。

原文发布时间为：2019-03-10
本文作者：丁威，《RocketMQ技术内幕》作者。
本文来自中间件兴趣圈，了解相关信息可以关注中间件兴趣圈。

Elasticsearch Bucket聚合(桶聚合) 第一篇(常用桶聚合一览)