关于后端:elasticsearch-之-histogram-直方图聚合

7次阅读

共计 7602 个字符，预计需要花费 20 分钟才能阅读完成。

直方图聚合 是一种基于多桶值聚合，可从文档中提取的 数值 或数值范畴值 来进行聚合。它能够对参加聚合的值来动静的生成固定大小的桶。

假如咱们有一个值是 32，并且桶的大小是5，那么 32 四舍五入后变成 30，因而文档将落入与键 30 关联的存储桶中。 上面的算式能够准确的确定每个文档的归属桶

bucket_key = Math.floor((value - offset) / interval) * interval + offset

offset：的值默认是从 0 开始。并且 offset 的值必须在 [0, interval) 之间。且须要是一个 负数。
value：值的参加计算的值，比方某个文档中的价格字段等。

此处是我本人的一个了解，如果谬误欢送指出。

存在的数据：[3, 8, 15]
offset = 0
interval = 5

那么可能会分成如下几个桶 [0,5) [5,10) [10, 15) [15,+∞)

数字 3 落入的桶 buket_key= Math.floor((3 - 0) / 5) * 5 + 0 = 0，即落入 [0,5) 这个桶中
数字 8 落入的桶 buket_key= Math.floor((8 - 0) / 5) * 5 + 0 = 5，即落入 [5,10) 这个桶中
数字 15 落入的桶 buket_key= Math.floor((15 - 0) / 5) * 5 + 0 = 15，即落入 [15,+∞) 这个桶中

咱们有一组 api 响应工夫数据，依据这组数据进行 histogram 聚合统计

PUT /index_api_response_time
{
  "settings": {"number_of_shards": 1},
  "mappings": {
    "properties": {
      "id": {"type": "long"},
      "api": {"type": "keyword"},
      "response_time": {"type": "integer"}
    }
  }
}

此处的 mapping 比较简单，就 3 个字段 id，api 和response_time。

PUT /index_api_response_time/_bulk
{"index":{"_id":1}}
{"api":"/user/infos","response_time": 3}
{"index":{"_id":2}}
{"api":"/user/add"}
{"index":{"_id":3}}
{"api":"/user/update","response_time": 8}
{"index":{"_id":4}}
{"api":"/user/list","response_time": 15}
{"index":{"_id":5}}
{"api":"/user/export","response_time": 30}
{"index":{"_id":6}}
{"api":"/user/detail","response_time": 32}

此处先记录 id=2的数据，这个是没有 response_time 的，前期聚合时额定解决。

GET /index_api_response_time/_search
{
  "size": 0,
  "aggs": {
    "agg_01": {
      "histogram": {
        "field": "response_time",
        "interval": 5
      }
    }
  }
}

@Test
@DisplayName("依据 response_time 聚合，距离为 5")
public void test01() throws IOException {
    SearchRequest request = SearchRequest.of(search ->
            search
                    .index("index_api_response_time")
                    .size(0)
                    .aggregations("agg_01", agg -> agg.histogram(histogram -> histogram.field("response_time")
                    .interval(5D))));
    System.out.println("request:" + request);
    SearchResponse<String> response = client.search(request, String.class);
    System.out.println("response:" + response);
}

此处聚合一下是为了联合已有的数据，看看每个数据是否落入到了相应的桶中

GET /index_api_response_time/_search
{
  "size": 0,

  "aggs": {
    "agg_01": {
      "histogram": {
        "field": "response_time",
        "interval": 5
      },
      "aggs": {
        "agg_sum": {
          "sum": {"field": "response_time"}
        }
      }
    }
  }
}

@Test
@DisplayName("在 test01 根底上聚合出每个桶总的响应工夫")
public void test02() throws IOException {
    SearchRequest request = SearchRequest.of(search ->
            search
                    .index("index_api_response_time")
                    .size(0)
                    .aggregations("agg_01", agg ->
                            agg.histogram(histogram -> histogram.field("response_time").interval(5D))
                               .aggregations("agg_sum", aggSum -> aggSum.sum(sum -> sum.field("response_time")))
                    ));
    System.out.println("request:" + request);
    SearchResponse<String> response = client.search(request, String.class);
    System.out.println("response:" + response);
}

从 5.1 中的后果咱们能够晓得，不论桶中是否存在数据，咱们都返回了，即返回了很多空桶。简略了解就是返回的桶中存在 doc_count=0 的数据，此处咱们须要将这个数据不返回

GET /index_api_response_time/_search
{
  "size": 0,

  "aggs": {
    "agg_01": {
      "histogram": {
        "field": "response_time",
        "interval": 5,
        "min_doc_count": 1
      }
    }
  }
}

@Test
@DisplayName("每个桶中必须存在 1 个文档的后果才返回 -min_doc_count")
public void test03() throws IOException {
    SearchRequest request = SearchRequest.of(search ->
            search
                    .index("index_api_response_time")
                    .size(0)
                    .aggregations("agg_01", agg -> agg.histogram(histogram -> histogram.field("response_time").interval(5D).minDocCount(1)
                            )
                    )
    );
    System.out.println("request:" + request);
    SearchResponse<String> response = client.search(request, String.class);
    System.out.println("response:" + response);
}

这个是什么意思？假如咱们通过 response_time >= 10 进行过滤，并且 interval=5 那么 es 默认状况下就不会返回 bucket_key =0,5,10 的桶，那么如果我想返回那么该如何解决呢？能够通过 extended_bounds 来实现 。
应用 extended_bounds 时，min_doc_count=0时才有意义。extended_bounds 不会过滤桶。

GET /index_api_response_time/_search
{
  "size": 0,
  "query": {
    "range": {
      "response_time": {"gte": 10}
    }
  }, 
  "aggs": {
    "agg_01": {
      "histogram": {
        "field": "response_time",
        "interval": 5,
        "min_doc_count": 0,
        "extended_bounds": {
          "min": 0,
          "max": 50
        }
      }
    }
  }
}

@Test
@DisplayName("补充空桶数据 -extended_bounds")
public void test04() throws IOException {
    SearchRequest request = SearchRequest.of(search ->
            search
                    .index("index_api_response_time")
                    .size(0)
                    .query(query-> query.range(range -> range.field("response_time").gte(JsonData.of(10))))
                    .aggregations("agg_01", agg -> agg.histogram(histogram -> histogram.field("response_time").interval(5D).minDocCount(0)
                                    .extendedBounds(bounds -> bounds.min(1D).max(50D))
                            )
                    )
    );
    System.out.println("request:" + request);
    SearchResponse<String> response = client.search(request, String.class);
    System.out.println("response:" + response);
}

此处的数据:

PUT /index_api_response_time/_bulk
{"index":{"_id":1}}
{"api":"/user/infos","response_time": 3}
{"index":{"_id":2}}
{"api":"/user/add"}
{"index":{"_id":3}}
{"api":"/user/update","response_time": 8}
{"index":{"_id":4}}
{"api":"/user/list","response_time": 15}
{"index":{"_id":5}}
{"api":"/user/export","response_time": 25}
{"index":{"_id":6}}
{"api":"/user/detail","response_time": 32}

GET /index_api_response_time/_search
{
  "size": 0,
  "query": {
    "range": {
      "response_time": {"gte": 10}
    }
  }, 
  "aggs": {
    "agg_01": {
      "histogram": {
        "field": "response_time",
        "interval": 5,
        "min_doc_count": 0,
        "hard_bounds": {
          "min": 15,
          "max": 25
        }
      },
      "aggs": {
        "a_s": {
          "sum": {"field": "response_time"}
        }
      }
    }
  }
}

@Test
@DisplayName("只展现 min-max 之间的桶 -hard_bounds")
public void test05() throws IOException {
    SearchRequest request = SearchRequest.of(search ->
            search
                    .index("index_api_response_time")
                    .size(0)
                    .query(query-> query.range(range -> range.field("response_time").gte(JsonData.of(10))))
                    .aggregations("agg_01", agg ->
                            agg.histogram(histogram -> histogram.field("response_time").interval(5D).minDocCount(0)
                                        .hardBounds(bounds -> bounds.min(1D).max(50D))
                            )
                               .aggregations("a_s", sumAgg -> sumAgg.sum(sum -> sum.field("response_time")))
                    )
    );
    System.out.println("request:" + request);
    SearchResponse<String> response = client.search(request, String.class);
    System.out.println("response:" + response);
}

By default the returned buckets are sorted by their key ascending, though the order behaviour can be controlled using the order setting. Supports the same order functionality as the Terms Aggregation.

GET /index_api_response_time/_search
{
  "size": 0,
  "query": {
    "range": {
      "response_time": {"gte": 10}
    }
  }, 
  "aggs": {
    "agg_01": {
      "histogram": {
        "field": "response_time",
        "interval": 5,
        "order": {"_count": "desc"}
      }
    }
  }
}

@Test
@DisplayName("排序 order")
public void test06() throws IOException {
    SearchRequest request = SearchRequest.of(search ->
            search
                    .index("index_api_response_time")
                    .size(0)
                    .query(query-> query.range(range -> range.field("response_time").gte(JsonData.of(10))))
                    .aggregations("agg_01", agg ->
                            agg.histogram(histogram -> histogram.field("response_time").interval(5D)
                                        .order(NamedValue.of("_count", SortOrder.Desc))
                            )
                    )
    );
    System.out.println("request:" + request);
    SearchResponse<String> response = client.search(request, String.class);
    System.out.println("response:" + response);
}

GET /index_api_response_time/_search
{
  "size": 0,
  "aggs": {
    "agg_01": {
      "histogram": {
        "field": "response_time",
        "interval": 5,
        "missing": 0
      }
    }
  }
}

@Test
@DisplayName("文档中缺失聚合字段时如何解决 -missing")
public void test07() throws IOException {
    SearchRequest request = SearchRequest.of(search ->
            search
                    .index("index_api_response_time")
                    .size(0)
                    .query(query-> query.range(range -> range.field("response_time").gte(JsonData.of(10))))
                    .aggregations("agg_01", agg ->
                            agg.histogram(histogram -> histogram.field("response_time").interval(5D) .missing(0D)
                            )
                    )
    );
    System.out.println("request:" + request);
    SearchResponse<String> response = client.search(request, String.class);
    System.out.println("response:" + response);
}

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

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

正文完

后端

发表至：后端

2022-11-25

0

关于后端:解析-RocketMQ-业务消息事务消息

关于后端:ELEC-279-面向对象详细教程

关于后端:项目终于用上了插入式注解真香

关于后端:好文推荐黑莓OS手册是如何详细阐述底层的进程和线程模型的

关于区块链:直播预告｜第七期红枣科技CEO月度DDC说明会

关于后端:elasticsearch-之-histogram-直方图聚合

1. 简介

2. bucket_key 如何计算

3. 有一组数据，如何确定是落入到那个桶中

4、需要

4.1 筹备 mapping

4.2 筹备数据

5、histogram 聚合操作

5.1、依据 response_time 聚合，距离为 5

5.1.1 dsl

5.1.2 java 代码

5.1.3 运行后果

5.2 在 5.1 根底上聚合出每个桶总的响应工夫

5.2.1 dsl

5.2.2 java 代码

5.2.3 运行后果

5.3 每个桶中必须存在 1 个文档的后果才返回 -min_doc_count

5.3.1 dsl

5.3.2 java 代码

5.3.3 运行后果

5.4 补充空桶数据 -extended_bounds

5.4.1 dsl

5.4.2 java 代码

5.4.3 运行后果

5.5 只展现 min-max 之间的桶 -hard_bounds

5.5.1 dsl

5.5.2 java 代码

5.5.3 运行后果

5.6 排序 -order

5.6.1 dsl

5.6.2 java 代码

5.6.3 运行后果

5.7 文档中缺失聚合字段时如何解决 -missing

5.7.1 dsl

5.7.2 java 代码

5.7.3 运行后果

6、残缺代码

7、参考文档

站内搜索