ElasticSearch-高级查询

前言

上文介绍了ES的各种查询，本文介绍如何在ES进行MySQL中的分组和聚合操作，并实现MySQL和ES之间的数据自动同步；
 
一、分组聚合

在ES中对于聚合查询，主要分为2大类：指标(Metric)聚合 与 桶(Bucket)聚合。

• 指标聚合：max、min、sum等，作用等同于Mysql中的相关聚合函数。
  
• 桶聚合：group by，作用等同于Mysql中根据哪1个字段进行分组
  

注意，我们不能对text类型的字段进行分组，因为text会进行分词，导致无法进行分组。
 
1.指标聚合（聚合函数）

指标聚合相当于MySQL中聚合函数，统计品牌为万豪的最贵酒店价格

1. GET /hotel/_search
2. {
3.   "query": {
4.     "term": {
5.       "brand": {
6.         "value": "万豪"
7.       }
8.     }
9.   },
10.   "size": 0,
11.   "aggs": {
12.     "最贵的": {
13.       "max": {
14.         "field": "price"
15.       }
16.     },
17.      "最便宜的": {
18.       "min": {
19.         "field": "price"
20.       }
21.     }
22.   }
23. }

复制代码

 
2.桶聚合（分组）

桶聚合相当于MySQL中的分组，统计品牌为万豪的酒店有哪些星级；

1. GET /hotel/_search
2. {
3.   "size": 0,
4.   "query": {
5.     "term": {
6.       "brand": {
7.         "value": "万豪"
8.       }
9.     }
10.   },
11.   "aggs": {
12.     "按星级名称分组": {
13.       "terms": {
14.         "field": "specs",
15.         "size": 20
16.       }
17.     }
18.   }
19.   
20. }

复制代码

 对数据库中所有数据，按照星级和品牌分组；

1. GET /hotel/_search
2. {
3.   "size": 0,
4.   "aggs": {
5.     "按品牌分组": {
6.       "terms": {
7.         "field": "brand",
8.         "size": 20
9.       }
10.     },
11.     "按星级分组": {
12.       "terms": {
13.         "field": "specs",
14.         "size": 20
15.       }
16.     }
17.   }
18. }

复制代码

3.总结

 在ES中1次请求，可以写多个聚合函数；
4.功能实现

根据搜索条件筛选之后，再根据品牌进行分组；
4.1.Kibana查询

根据搜索条件对品牌进行数据分组

1. GET hotel/_search
2. {
3.   "size": 0,
4.   "query": {
5.     "query_string": {
6.       "fields": ["name","synopsis","area","address"],
7.       "query": "三亚 OR 商务"
8.     }
9.   },
10.   "aggs": {
11.     "hotel_brands": {
12.       "terms": {
13.         "field": "brand",
14.         "size": 100
15.       }
16.     }
17.   }
18. }

复制代码

4.2.JavaAPI查询

1. @Override
2.     public Map<String, Object> searchBrandGroupQuery(Integer current, Integer size, Map<String, Object> searchParam) {
3.         //设置查询请求头
4.         SearchRequest searchRequest = new SearchRequest("hotel");
5.         //设置查询请求体
6.         SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
7.         //设置查询方式
8.         if (!StringUtils.isEmpty(searchParam.get("condition"))) {
9.             QueryBuilder queryBuilder = QueryBuilders.queryStringQuery(searchParam.get("condition").toString())
10.                     .field("name")
11.                     .field("synopsis")
12.                     .field("area")
13.                     .field("address")
14.                     .defaultOperator(Operator.OR);
15.             searchSourceBuilder.query(queryBuilder);
16.         }
17.         //设置按品牌分组
18.         AggregationBuilder aggregationBuilder = AggregationBuilders.terms("brand_groups")
19.                 .size(200)
20.                 .field("brand");
21.         searchSourceBuilder.aggregation(aggregationBuilder);
23.         //设置分页
24.         searchSourceBuilder.from((current - 1) * size);
25.         searchSourceBuilder.size(size);
26.         searchRequest.source(searchSourceBuilder);
27.         try {
28.             SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
29.             SearchHits hits = searchResponse.getHits();
30.             long totalHits = hits.getTotalHits().value;
31.             ArrayList<String> groupNameList = new ArrayList<>();
32.             //获取并处理聚合查询结果
33.             Terms brandGroups = searchResponse.getAggregations().get("brand_groups");
34.             for (Terms.Bucket bucket : brandGroups.getBuckets()) {
35.                 String key = (String) bucket.getKey();
36.                 groupNameList.add(key);
37.             }
39.             Map<String, Object> map = new HashMap<>();
40. //            map.put("list", list);
41.             map.put("totalResultSize", totalHits);
42.             map.put("current", current);
43.             //设置总页数
44.             map.put("totalPage", (totalHits + size - 1) / size);
45.             //设置品牌分组列表
46.             map.put("brandList", groupNameList);
47.             return map;
48.         } catch (IOException e) {
49.             e.printStackTrace();
50.         }
51.         return null;
52.     }

复制代码

HotelServiceImpl.java 
5.分组和聚合一起使用

通常情况我们统计数据时，会先进行分组，然后再在分组的基础上进行聚合操作；
根据用户输入的日期，统计某品牌下所有酒店销量。 对于该功能的实现，需要进行多层聚合。

• 对日期时间段范围查询
  
• 根据品牌进行分组查询
  
• 对分组查询结果进行sum聚合
  

5.1.Kibana查询

在桶聚合中嵌套指标聚合，就等于MySQL会先进行分组操作，然后再在数据分组的基础上，进行聚合函数统计；

1. GET hotel/_search
2. {
3.   "size": 0,
4.   "query": {
5.     "range": {
6.       "createTime": {
7.         "gte": "2015-01-01",
8.         "lte": "2015-12-31"
9.       }
10.     }
11.   },
12.   "aggs": {
13.     "根据品牌分组": {
14.       "terms": {
15.         "field": "brand",
16.         "size": 100
17.       },
18.       "aggs": {
19.         "该品牌总销量": {
20.           "sum": {
21.             "field": "salesVolume"
22.           }
23.         },
24.         "该品牌销量平均值": {
25.           "avg": {
26.             "field": "salesVolume"
27.           }
28.         }
29.       }
30.     }
31.   }
32. }

复制代码

5.2.JavaAPI查询

[code]public List searchDateHistogram(Map searchParam) {    //定义结果集    List result = new ArrayList();    //设置查询    SearchRequest searchRequest = new SearchRequest("hotel");    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();    //todo 自定义日期时间段范围查询    RangeQueryBuilder queryBuilder = QueryBuilders.rangeQuery("createTime")        .gte(searchParam.get("minTime"))        .lte(searchParam.get("maxTime"))        .format("yyyy-MM-dd");    searchSourceBuilder.query(queryBuilder);    //todo 聚合查询设置    TermsAggregationBuilder aggregationBuilder = AggregationBuilders.terms("hotel_brand").field("brand").size(100);    //构建二级聚合    SumAggregationBuilder secondAggregation = AggregationBuilders.sum("hotel_salesVolume").field("salesVolume");    aggregationBuilder.subAggregation(secondAggregation);    searchSourceBuilder.aggregation(aggregationBuilder);    searchRequest.source(searchSourceBuilder);    try {        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);        //todo 获取聚合结果并处理        Aggregations aggregations = searchResponse.getAggregations();        Map aggregationMap = aggregations.asMap();        Terms terms = (Terms) aggregationMap.get("hotel_brand");        List