ES度量聚合(ElasticSearch Metric Aggregations)
微信公众号：[中间件兴趣圈]
作者简介：《RocketMQ技术内幕》作者；
从本篇将开始进入ES系列的聚合部分(Aggregations)。
本篇重点介绍Elasticsearch Metric Aggregations(度量聚合)。
Metric聚合，主要针对数值类型的字段，类似于关系型数据库中的sum、avg、max、min等聚合类型。
本例基于如下索引进行试验：
1public static void createMapping_agregations() {
2 RestHighLevelClient client = EsClient.getClient();
3 try {
4 CreateIndexRequest request = new CreateIndexRequest("aggregations_index02");
5 XContentBuilder jsonBuilder = XContentFactory.jsonBuilder()
6 .startObject()
7 .startObject("properties")
8 .startObject("orderId")
9 .field("type", "integer")
10 .endObject()
11 .startObject("orderNo")
12 .field("type", "keyword")
13 .endObject()
14 .startObject("totalPrice")
15 .field("type", "double")
16 .endObject()
17 .startObject("sellerId")
18 .field("type", "integer")
19 .endObject()
20 .startObject("sellerName")
21 .field("type", "keyword")
22 .endObject()
23 .startObject("buyerId")
24 .field("type", "integer")
25 .endObject()
26 .startObject("buyerName")
27 .field("type", "keyword")
28 .endObject()
29 .startObject("createTime")
30 .field("type", "date")
31 .field("format", "yyyy-MM-dd HH:mm:ss")
32 .endObject()
33 .startObject("status")
34 .field("type", "integer")
35 .endObject()
36 .startObject("reciveAddressId")
37 .field("type", "integer")
38 .endObject()
39 .startObject("reciveName")
40 .field("type", "keyword")
41 .endObject()
42 .startObject("phone")
43 .field("type", "keyword")
44 .endObject()
45 .startObject("skuId")
46 .field("type", "integer")
47 .endObject()
48 .startObject("skuNo")
49 .field("type", "keyword")
50 .endObject()
51 .startObject("goodsId")
52 .field("type", "integer")
53 .endObject()
54 .startObject("goodsName")
55 .field("type", "keyword")
56 .endObject()
57 .startObject("num")
58 .field("type", "integer")
59 .endObject()
60 .endObject()
61 .endObject();
62 request.mapping("_doc", jsonBuilder);
63 System.out.println(client.indices().create(request, RequestOptions.DEFAULT));
64 } catch (Throwable e) {
65 e.printStackTrace();
66 } finally {
67 EsClient.close(client);
68 }
69 }
对应的SQL表结构如下：
1CREATE TABLE `es_order_tmp` (
2 `orderId` int(11) NOT NULL DEFAULT '0' COMMENT '主键',
3 `orderNo` varchar(30) DEFAULT NULL COMMENT '订单编号',
4 `totalPrice` decimal(10,2) DEFAULT NULL COMMENT '订单总价，跟支付中心返回金额相等，包括了雅豆，余额，第三方支付的金额。运费包含在内，优惠券抵扣的金额不含在内',
5 `sellerId` int(11) DEFAULT NULL COMMENT '商家ID',
6 `selerName` varchar(50) DEFAULT NULL COMMENT '商家名称',
7 `buyerId` int(11) DEFAULT NULL COMMENT '创建者，购买者',
8 `buyerName` varchar(255) DEFAULT NULL COMMENT '业主姓名',
9 `createTime` varchar(22) DEFAULT NULL,
10 `status` int(11) DEFAULT NULL COMMENT '订单状态,0:待付款,1:待发货,2:待收货,3:待评价,4:订单完成,5:订单取消,6:退款处理中,7:拒绝退货,8:同意退货,9:退款成功,10:退款关闭,11:订单支付超时,12:半支付状态',
11 `reciveAddressId` int(11) DEFAULT NULL COMMENT '收货地址ID',
12 `reciveName` varchar(50) DEFAULT NULL,
13 `phone` varchar(30) DEFAULT NULL COMMENT '联系号码',
14 `skuId` int(11) DEFAULT NULL COMMENT '货品ID',
15 `skuNo` varchar(100) DEFAULT NULL COMMENT 'SKU编号',
16 `goodsId` int(11) DEFAULT NULL COMMENT '商品ID',
17 `goodsName` varchar(100) DEFAULT NULL COMMENT '商品名称',
18 `num` int(11) DEFAULT NULL COMMENT '数量'
19) ENGINE=InnoDB DEFAULT CHARSET=utf8;
平均值聚合
平均值聚合。
注：max,sum,min等使用与avg类似，故不重复介绍。
1POST /exams/_search?size=0
2{
3 "aggs" : {
4 "avg_grade" : { "avg" : { "field" : "grade" } }
5 }
6}
对字段grade取平均值。
对应的java示例如下：
1public static void testMatchQuery() {
2 RestHighLevelClient client = EsClient.getClient();
3 try {
4 SearchRequest searchRequest = new SearchRequest();
5 searchRequest.indices("aggregations_index02");
6 SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
7 AggregationBuilder avg = AggregationBuilders.avg("avg-aggregation").field("num").missing(0); // @1
8 sourceBuilder.aggregation(avg);
9 sourceBuilder.size(0);
10 sourceBuilder.query(
11 QueryBuilders.termQuery("sellerId", 24)
12 );
13 searchRequest.source(sourceBuilder);
14 SearchResponse result = client.search(searchRequest, RequestOptions.DEFAULT);
15 System.out.println(result);
16 } catch (Throwable e) {
17 e.printStackTrace();
18 } finally {
19 EsClient.close(client);
20 }
21 }
其中代码@1：missing(0)表示如果文档中没有取平均值的字段时，则使用该值进行计算，本例中使用0参与计算。
其返回结果如下：
1{
2 "took":2,
3 "timed_out":false,
4 "_shards":{
5 "total":5,
6 "successful":5,
7 "skipped":0,
8 "failed":0
9 },
10 "hits":{
11 "total":39,
12 "max_score":0,
13 "hits":[
14
15 ]
16 },
17 "aggregations":{
18 "avg#avg-aggregation":{
19 "value":1.2820512820512822
20 }
21 }
22}
加权平均聚合
加权平均聚合，其算法，∑(value * weight) / ∑(weight)。
加权平均（weghted_avg）支持的参数列表：
value
提供值的字段或脚本的配置。例如定义计算哪个字段的平均值，该值支持如下子参数：field
用来定义平均值的字段名称。missing
用来定义如果匹配到的文档没有avg字段，使用该值来参与计算。weight
用来定义权重的对象，其可选属性如下：field
定义权重来源的字段。missing
如果文档缺失权重来源字段，以该值来代表该文档的权重值。format
数值类型格式化。value_type
用来指定value的类型，例如ValueType.DATE、ValueType.IP等。
示例如下：
1POST /exams/_search
2{
3 "size": 0,
4 "aggs" : {
5 "weighted_grade": {
6 "weighted_avg": {
7 "value": {
8 "field": "grade"
9 },
10 "weight": {
11 "field": "weight" // @2
12 }
13 }
14 }
15 }
16}
从文档中抽取属性为weight的字段的值来当权重值。
其JAVA示例如下：
1public static void test_weight_avg_aggregation() {
2 RestHighLevelClient client = EsClient.getClient();
3 try {
4 SearchRequest searchRequest = new SearchRequest();
5 searchRequest.indices("aggregations_index02");
6 SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
7 WeightedAvgAggregationBuilder avg = AggregationBuilders.weightedAvg("avg-aggregation")
8
9 .value(
10 (new MultiValuesSourceFieldConfig.Builder())
11 .setFieldName("num")
12 .setMissing(0)
13 .build()
14 )
15 .weight(
16 (new MultiValuesSourceFieldConfig.Builder())
17 .setFieldName("num")
18 .setMissing(1)
19 .build()
20 )
21 // .valueType(ValueType.LONG)
22
23 ;
24
25 avg.toString();
26
27 sourceBuilder.aggregation(avg);
28 sourceBuilder.size(0);
29 sourceBuilder.query(
30 QueryBuilders.termQuery("sellerId", 24)
31 );
32 searchRequest.source(sourceBuilder);
33 SearchResponse result = client.search(searchRequest, RequestOptions.DEFAULT);
34 System.out.println(result);
35 } catch (Throwable e) {
36 e.printStackTrace();
37 } finally {
38 EsClient.close(client);
39 }
40 }
基数聚合
基数聚合，先distinct,再聚合，类似关系型数据库(count(distinct))。
示例如下：
1POST /sales/_search?size=0
2{
3 "aggs" : {
4 "type_count" : {
5 "cardinality" : {
6 "field" : "type"
7 }
8 }
9 }
10}
对应的JAVA示例如下：
1public static void test_Cardinality_Aggregation() {
2 RestHighLevelClient client = EsClient.getClient();
3 try {
4 SearchRequest searchRequest = new SearchRequest();
5 searchRequest.indices("aggregations_index02");
6 SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
7 AggregationBuilder aggregationBuild = AggregationBuilders.cardinality("buyerid_count").field("buyerId");
8 sourceBuilder.aggregation(aggregationBuild);
9 sourceBuilder.size(0);
10 sourceBuilder.query(
11 QueryBuilders.termQuery("sellerId", 24)
12 );
13 searchRequest.source(sourceBuilder);
14 SearchResponse result = client.search(searchRequest, RequestOptions.DEFAULT);
15 System.out.println(result);
16 } catch (Throwable e) {
17 e.printStackTrace();
18 } finally {
19 EsClient.close(client);
20 }
21 }
返回结果如下：
1{
2 "took":30,
3 "timed_out":false,
4 "_shards":{
5 "total":5,
6 "successful":5,
7 "skipped":0,
8 "failed":0
9 },
10 "hits":{
11 "total":39,
12 "max_score":0,
13 "hits":[
14
15 ]
16 },
17 "aggregations":{
18 "cardinality#type_count":{
19 "value":11
20 }
21 }
22}
上述实现与SQL:SELECT COUNT(DISTINCT buyerId) from es_order_tmp where sellerId=24; 效果类似，表示购买了商家id为24的买家个数。
其核心参数如下：
precision_threshold
精确度控制。在此计数之下，期望计数接近准确。在这个值之上，计数可能会变得更加模糊（不准确）。支持的最大值是40000，超过此值的阈值与40000的阈值具有相同的效果。默认值是3000。
上述示例中返回的11是精确值，如果改写成下面的代码，结果将变的不准确：
1field("buyerId").precisionThreshold(5)
其返回结果如下：
1{
2 "took":5,
3 "timed_out":false,
4 "_shards":{
5 "total":5,
6 "successful":5,
7 "skipped":0,
8 "failed":0
9 },
10 "hits":{
11 "total":39,
12 "max_score":0,
13 "hits":[
14
15 ]
16 },
17 "aggregations":{
18 "cardinality#buyerid_count":{
19 "value":9
20 }
21 }
22}
Pre-computed hashes
一个比较好的实践是需要对字符串类型的字段进行基数聚合的话，可以提前索引该字符串的hash值，通过对hash值的聚合，提高效率。Missing Value
missing参数定义了应该如何处理缺少值的文档。默认情况下，它们将被忽略，但也可以将它们视为具有一个值，通过missing value来设置。
中位绝对偏差聚合
中位绝对偏差聚合。由于这部分内容与统计学关系密切，但这并不是我的特长，故对该统计的含义做深入解读，在实际场景中，我们只需要知道ES提供了中位数偏差统计的功能，如果有这方面的需求，我们知道如何使用ES的中位数统计即可。
官方场景：
假设我们收集了商品评价数据（1星到5星之间的数值）。在实际使用过程中通常会使用平均值来展示商品的整体评价等级。中位绝对偏差聚合可以帮助我们了解评审之间的差异有多大。
在这个例子中，我们有一个平均评级为3星的产品。让我们看看它的评级的绝对偏差中值，以确定它们的变化有多大。按照我的理解，中位绝对偏差聚合 ，聚合的数据来源于（原始数据 - 所有原始数值的平均值 的绝对值进行聚合）。
例如评论原始数据如下：
1、2、5、5、4、3、5、5、5、5
其平均值：4
那中位数绝对偏差值聚合的数据为：
3、2、1、1、0、1、1、1、1、1
其Restfull示例如下：
1GET reviews/_search
2{
3 "size": 0,
4 "aggs": {
5 "review_average": { // @1
6 "avg": {
7 "field": "rating"
8 }
9 },
10 "review_variability": { // @2
11 "median_absolute_deviation": {
12 "field": "rating"
13 }
14 }
15 }
16}
该聚合包含两部分。
代码@1：针对字段rating使用AVG进行聚合（平均聚合，求出中位数）
代码@2：针对字段rating进行中位数绝对偏差聚合。
备注：在es high rest api中未封装(median absolute deviation aggregation)聚合。
ES 关于 Metric聚合就介绍到这里了，接下来将重点分析Es Buket聚合。
关于作者：《RocketMQ计算内幕：RocketMQ架构设计与实现原理》。
《RocketMQ技术内幕》已出版上市，目前可在主流购物平台（京东、天猫等）购买，本书从源码角度深度分析了RocketMQ NameServer、消息发送、消息存储、消息消费、消息过滤、主从同步HA、事务消息；在实战篇重点介绍了RocketMQ运维管理界面与当前支持的39个运维命令；并在附录部分罗列了RocketMQ几乎所有的配置参数。本书得到了RocketMQ创始人、阿里巴巴Messaging开源技术负责人、Linux OpenMessaging 主席的高度认可并作序推荐。目前是国内第一本成体系剖析RocketMQ的书籍。
《RocketMQ技术内幕》宣传海报：
更多文章请关注微信公众号：