Elasticsearch Bucket聚合(桶聚合) 第一篇(常用桶聚合一览)-低调大师

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

2019-09-02 756

本篇将开始介绍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技术内幕》作者。
本文来自中间件兴趣圈，了解相关信息可以关注中间件兴趣圈。

微信关注我们

原文链接：https://yq.aliyun.com/articles/717151

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

ES度量聚合(ElasticSearch Metric Aggregations)

从本篇将开始进入ES系列的聚合部分(Aggregations)。本篇重点介绍Elasticsearch Metric Aggregations(度量聚合)。 Metric聚合，主要针对数值类型的字段，类似于关系型数据库中的sum、avg、max、min等聚合类型。本例基于如下索引进行试验： public static void createMapping_agregations() { RestHighLevelClient client = EsClient.getClient(); try { CreateIndexRequest request = new CreateIndexRequest("aggregations_index02"); XContentBuilder jsonBuilder = XContentFactory.jsonBuilder() .startObject() .startObject("properties") .startObject("orderId") .field("type", "integer") .endObject() .sta...

2019-09-02

707

概述本文主要介绍X-Pack Spark集群监控指标的查看方法。Spark集群对接了Ganglia和云监控。下面分别介绍两者的使用方法。 Ganglia Ganglia是一个分布式监控系统。 Ganglia 入口打开Spark集群依次进入：数据库连接>UI访问>详细监控UI>Ganglia。如下图：注意：打开之前请先阅读UI访问说明。 Ganglia 界面介绍本只做常用的介绍。如下图：导航栏选择不同的功能查看，本文主要介绍常用的“Main” 时间选择选择查看不同的时间段的资源使用情况。统计信息统计信息主要列出集群的资源总体使用情况。每个字段解释如下表: 名称解释 CPUs Total 集群的CPU总数 Hosts up 集群的总节点数：正在运行的节点 Hosts down 集群的总节点数：停止运行的节点 Current Load Avg

2019-09-04

719

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。