Elasticsearch学习总结八 ElasticSearch中的聚合操作

首先准备数据，索引包含四个字段fieldA,fieldB,fieldC,fieldD,如下图，以下案列中都使用了基本REST命令和JavaAP两种方式实现

输入图片说明

1). 首先按照某个字段fieldC分组统计,相当于sql 中的group by操作,

1curl -XPOST "http://121.40.128.155:9200/tempindex/_search?pretty" -d '{
2"size": 0,
3  "aggs": {
4    "fieldC_count": {
5      "terms": {
6        "field": "fieldC"
7      }
8    }
9  }
10}'
11
12返回值如下：
13{
14  "took" : 3,
15  "timed_out" : false,
16  "_shards" : {
17    "total" : 5,
18    "successful" : 5,
19    "failed" : 0
20  },
21  "hits" : {
22    "total" : 5,
23    "max_score" : 0.0,
24    "hits" : [ ]
25  },
26  "aggregations" : {
27    "fieldC_count" : {
28      "doc_count_error_upper_bound" : 0,
29      "sum_other_doc_count" : 0,
30      "buckets" : [ {
31        "key" : "java",
32        "doc_count" : 3
33      }, {
34        "key" : "c++",
35        "doc_count" : 1
36      }, {
37        "key" : "ptyhone",
38        "doc_count" : 1
39      } ]
40    }
41  }
42}

对应的JavaApi是如何实现的呢

1EsSearchManager esSearchManager = EsSearchManager.getInstance();
2SearchRequestBuilder searchReq = esSearchManager.client.prepareSearch("tempindex");
3searchReq.setTypes("tempindex");
4//group by 条件
5TermsBuilder termsb = AggregationBuilders.terms("my_fieldC").field("fieldC").size(100);
6searchReq.addAggregation(termsb);
7SearchResponse searchRes = searchReq.execute().actionGet();
8Terms fieldATerms = searchRes.getAggregations().get("my_fieldC");
9for (Terms.Bucket filedABucket : fieldATerms.getBuckets()) {
10	//fieldA
11	String groupbyKey = filedABucket.getKey().toString();
12	//COUNT(fieldA)
13	long  countValue = filedABucket.getDocCount();
14}
15

2).统计某一个字段的最大最小值

1curl -XPOST "http://121.40.128.155:9200/tempindex/_search?pretty" -d '{
2  "size": 0,
3  "aggs": {
4    "max_fieldA": {
5      "max": {
6        "field": "fieldA"
7      }
8    },
9    "min_fieldA": {
10      "min": {
11        "field": "fieldA"
12      }
13    }
14  }
15}'
16
17返回值如下：
18{
19  "took" : 10,
20  "timed_out" : false,
21  "_shards" : {
22    "total" : 5,
23    "successful" : 5,
24    "failed" : 0
25  },
26  "hits" : {
27    "total" : 5,
28    "max_score" : 0.0,
29    "hits" : [ ]
30  },
31  "aggregations" : {
32    "max_fieldA" : {
33      "value" : 25.0
34    },
35    "min_fieldA" : {
36      "value" : 10.0
37    }
38  }
39}

对应的JavaApi实现如下

1EsSearchManager esSearchManager = EsSearchManager.getInstance();
2SearchRequestBuilder searchReq = esSearchManager.client.prepareSearch("tempindex");
3searchReq.setTypes("tempindex");
4MaxBuilder maxBuilder = AggregationBuilders.max("max_fieldA").field("fieldA");
5searchReq.addAggregation(maxBuilder);
6MinBuilder minBuilder = AggregationBuilders.min("min_fieldA").field("fieldA");
7searchReq.addAggregation(minBuilder);
8SearchResponse searchRes = searchReq.execute().actionGet();
9InternalMax internalMax=searchRes.getAggregations().get("max_fieldA");
10System.out.println(internalMax.getName() +"="+ internalMax.getValue());
11InternalMin internalMin=searchRes.getAggregations().get("min_fieldA");
12System.out.println(internalMin.getName() +"="+ internalMin.getValue());
13

3).Average平均值按照某个字段求平均值

1curl -XPOST "http://121.40.128.155:9200/tempindex/_search?pretty" -d '{
2"size": 0,
3"aggs": {
4    "per_count": {
5        "terms": {
6            "field": "fieldC"
7        },
8        "aggs": {
9            "avg_fieldB": {
10                "avg": {
11                    "field": "fieldB"
12                }
13            }
14        }
15    }
16}
17}'	
18
19返回值如下：
20{
21  "took" : 3,
22  "timed_out" : false,
23  "_shards" : {
24    "total" : 5,
25    "successful" : 5,
26    "failed" : 0
27  },
28  "hits" : {
29    "total" : 5,
30    "max_score" : 0.0,
31    "hits" : [ ]
32  },
33  "aggregations" : {
34    "per_count" : {
35      "doc_count_error_upper_bound" : 0,
36      "sum_other_doc_count" : 0,
37      "buckets" : [ {
38        "key" : "java",
39        "doc_count" : 3,
40        "avg_fieldB" : {
41          "value" : 20.0
42        }
43      }, {
44        "key" : "c++",
45        "doc_count" : 1,
46        "avg_fieldB" : {
47          "value" : 35.0
48        }
49      }, {
50        "key" : "ptyhone",
51        "doc_count" : 1,
52        "avg_fieldB" : {
53          "value" : 25.0
54        }
55      } ]
56    }
57  }
58}
59

对应的JavaApi实现如下

1EsSearchManager esSearchManager = EsSearchManager.getInstance();
2SearchRequestBuilder searchReq = esSearchManager.client.prepareSearch("tempindex");
3searchReq.setTypes("tempindex");
4TermsBuilder termsb = AggregationBuilders.terms("my_fieldC").field("fieldC").size(100);
5termsb.subAggregation(AggregationBuilders.avg("my_avg_fieldB").field("fieldB"));
6searchReq.setQuery(QueryBuilders.matchAllQuery()).addAggregation(termsb);
7SearchResponse searchRes = searchReq.execute().actionGet();
8Terms fieldATerms = searchRes.getAggregations().get("my_fieldC");
9for (Terms.Bucket filedABucket : fieldATerms.getBuckets()) {
10	String fieldAValue = filedABucket.getKey().toString();
11	long fieldACount = filedABucket.getDocCount();
12	Avg avgagg = filedABucket.getAggregations().get("my_avg_fieldB");
13	double avgFieldB = avgagg.getValue();
14	System.out.println("fieldAValue="+fieldAValue);
15	System.out.println("fieldACount="+fieldACount);
16	System.out.println("avgFieldB="+avgFieldB);
17}
18

4).Sum求和,求某个字段的sum之和

1curl -XPOST "http://121.40.128.155:9200/tempindex/_search?pretty" -d '{
2"size": 0,
3"aggs": {
4    "per_count": {
5        "terms": {
6            "field": "fieldC"
7        },
8        "aggs": {
9            "sum_fieldB": {
10                "sum": {
11                    "field": "fieldB"
12                }
13            }
14        }
15    }
16}
17}'
18返回值如下：
19{
20  "took" : 2,
21  "timed_out" : false,
22  "_shards" : {
23    "total" : 5,
24    "successful" : 5,
25    "failed" : 0
26  },
27  "hits" : {
28    "total" : 5,
29    "max_score" : 0.0,
30    "hits" : [ ]
31  },
32  "aggregations" : {
33    "per_count" : {
34      "doc_count_error_upper_bound" : 0,
35      "sum_other_doc_count" : 0,
36      "buckets" : [ {
37        "key" : "java",
38        "doc_count" : 3,
39        "sum_fieldB" : {
40          "value" : 60.0
41        }
42      }, {
43        "key" : "c++",
44        "doc_count" : 1,
45        "sum_fieldB" : {
46          "value" : 35.0
47        }
48      }, {
49        "key" : "ptyhone",
50        "doc_count" : 1,
51        "sum_fieldB" : {
52          "value" : 25.0
53        }
54      } ]
55    }
56  }

对应的Java代码如下

1EsSearchManager esSearchManager = EsSearchManager.getInstance();
2SearchRequestBuilder searchReq = esSearchManager.client.prepareSearch("tempindex");
3searchReq.setTypes("tempindex");
4TermsBuilder termsb = AggregationBuilders.terms("my_fieldC").field("fieldC").size(100);
5termsb.subAggregation(AggregationBuilders.sum("my_sum_fieldB").field("fieldB"));
6searchReq.setQuery(QueryBuilders.matchAllQuery()).addAggregation(termsb);
7SearchResponse searchRes = searchReq.execute().actionGet();
8Terms fieldATerms = searchRes.getAggregations().get("my_fieldC");
9for (Terms.Bucket filedABucket : fieldATerms.getBuckets()) {
10	String fieldAValue = filedABucket.getKey().toString();
11	long fieldACount = filedABucket.getDocCount();
12	Avg avgagg = filedABucket.getAggregations().get("my_sum_fieldB");
13	double avgFieldB = avgagg.getValue();
14	System.out.println("fieldAValue="+fieldAValue);
15	System.out.println("fieldACount="+fieldACount);
16	System.out.println("avgFieldB="+avgFieldB);
17}

以上总结了部分es基本操作API和调用demo，详细的代码请查看github地址 https://github.com/winstonelei/BigDataTools

Elasticsearch学习总结八 ElasticSearch中的聚合操作

首先准备数据，索引包含四个字段fieldA,fieldB,fieldC,fieldD,如下图，以下案列中都使用了基本REST命令和JavaAP两种方式实现

评论区

相关推荐