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。

其取值：

1. BucketOrder.count(boolean asc)
   按匹配文档格式升序/降序排序。
2. BucketOrder.key(boolean asc)
   按key的升序或降序排序。
3. BucketOrder.aggregation
   通过定义一个子聚合进行排序。
4. 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. 1小时(1h)是指定时区内第一个小时的00:00分钟到下一个小时的00:00分钟之间的时间间隔，用来补偿其间的任何闰秒，从而使经过该小时的分钟数和秒数在开始和结束时相同。
2. {n}h，表示时间间隔，等于 n 60 60 * 1000 毫秒的时间间隔。

• days (d)

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

• weeks (w)

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

• months (M)

1. 一个月(1M)是本月开始之间的时间间隔的一天与次月的同一天。
2. 不支持{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技术内幕》作者。
本文来自 中间件兴趣圈，了解相关信息可以关注 中间件兴趣圈。