javaAPI实现elasticsearch5.5.2的聚合分析
·
前言：
在前面的几篇文章讲到了elasticsearch的搜索，但是elasticsearch还有强大的聚合分析功能，通过聚合，我们会得到一个数据的概览，这样对大数据提取统计指标时就变得游刃有余。聚合允许我们向数据提出一些复杂的问题。虽然功能完全不同于搜索，但它们使用相同的数据结构，执行速度很快并且就像搜索一样几乎是实时的。
一、聚合（Aggregations）的介绍
elasticsearch的聚合（Aggregations）类似于数据库sql中的分组 group by，count、sum等函数，除此之外他还有更多强大的数据统计分析接口。
聚合有两个核心概念：
1.桶（bucket）：对数据进行分组。比如一个对象User他有一个属性是city，有如下数据：1.张三 上海 ；2.李四 北京； 3.王五 北京，我们就可以基于city划分buckets，一个是北京bucket，一个是上海bucket，按照某个字段进行bucket划分，那个字段的值相同的那些数据，就会被划分到一个bucket中，相当于sql中的group by分组。
2.指标（metric）：对一个数据分组执行的统计。当我们有了一堆bucket之后，就可以对每个bucket中的数据进行聚合分词了，比如说计算一个bucket内所有数据的数量、平均值、最大值等的，metric就是对一个bucket执行的某种聚合分析的操作，相当于数据库中的avg,sum函数
二、Bucket aggregations 聚合分组的使用
官方参考资料：
https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/_bucket_aggregations.html
我会在第一个例子中将完整的代码写出来，其他的例子仿照第一个进行
1. Terms Aggregation
terms按照某个字段进行分组，下面的代码中也涉及了排序问题，详细代码看注释
/**TermAggs 包含了Order
* termAggs统计每个颜色的个数
* @Title: countByColor
* @Description: TODO(统计每个颜色的销量)
* @param @throws UnknownHostException
* @return void
* @autor lpl
* @date 2017年11月6日
* @throws
*/
@Test
public void countByColor() throws UnknownHostException{
//进行聚合查询,terms按照某个字段中进行分组
TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders
.terms("popular_colors") //aggs的名称，自定义，取数据的时候与其一致即可
.field("color") //需要分组的字段
//.order(Terms.Order.term(true)) //true表示按照term分组的asc排序
.order(Terms.Order.count(true)); //true表asc按照doc_count排序，false表示desc排序
//查询数据进行聚合分析
SearchResponse response = EsClient.client().prepareSearch("tvs").setTypes("sales")
.addAggregation(termsAggregationBuilder)
.setSize(0)
.execute()
.actionGet();
//获取聚合
Terms terms = response.getAggregations().get("popular_colors");
//遍历处理数据
for (Terms.Bucket entry : terms.getBuckets()) {
Map<String,Object> map = new HashMap<String,Object>();
String key = (String) entry.getKey(); // Term
long docCount = entry.getDocCount(); // Doc count
//将结果放到Map中
map.put("color",key);
map.put("docCount",docCount);
System.out.println(map);
}
}在Terms Aggregation涉及到一个Order分组问题，在上面的代码上我已经做了简单的注释，但是还存在一种排序方式，就是我们按照某一个统计出来字段进行排序，比如：按照height字段的平均值进行排序，fasle代表了desc排序
AggregationBuilders
.terms("genders")
.field("gender")
.order(Terms.Order.aggregation("avg_height", false))
.subAggregation(
AggregationBuilders.avg("avg_height").field("height")
)2. Histogram Aggregation 按照区间进行分组
按照区间分组有Histogram Aggregation 和 Date Histogram Aggregation，我们以两个简单的需求进行编写具体的代码
2.1Histogram Aggregation：以2000位区间单位统计电视的销量
/**
* 以2000位区间单位统计电视的销量
* @Title: HistogramTest
* @param @throws UnknownHostException
* @return void
* @autor lpl
* @date 2017年11月6日
* @throws
*/
@Test
public void HistogramTest() throws UnknownHostException{
//histogram：类似于terms，也是进行bucket分组操作,interval：2000，
//划分范围，0~2000，2000~4000，4000~6000，6000~8000，8000~10000
HistogramAggregationBuilder histogramAggregationBuilder = AggregationBuilders
.histogram("his_price")
.field("price")
.interval(2000);
//查询数据进行聚合分析
SearchResponse response = EsClient.client()
.prepareSearch("tvs").setTypes("sales")
.addAggregation(histogramAggregationBuilder)
.setSize(0)
.execute()
.actionGet();
Histogram agg = response.getAggregations().get("his_price");
// For each entry
for (Histogram.Bucket entry : agg.getBuckets()) {
Map<String,Object> map = new HashMap<String,Object>();
Number key = (Number) entry.getKey(); // 区间key
long docCount = entry.getDocCount(); // Doc count
//数据封装
map.put("key",key);
map.put("doc_count",docCount);
System.out.println(map);
}
}/**
* 统计每个月份的电视销量
* @Title: dateHistogramTest
* @Description: TODO(这里用一句话描述这个方法的作用)
* @param @throws UnknownHostException
* @return void
* @autor lpl
* @date 2017年11月6日
* @throws
*/
@Test
public void dateHistogramTest() throws UnknownHostException{
//Date histogram的用法与histogram差不多，只不过区间上支持了日期的表达式。
DateHistogramAggregationBuilder dateHistogramInterval = AggregationBuilders
.dateHistogram("date_hist") //聚合的名称，可以随意去取，下边取数据一致即可
.field("sold_date") //需要进行分组的字段
.dateHistogramInterval(DateHistogramInterval.MONTH) //设置时间区间
.minDocCount(0) //设置最小的 doc_count，
.format("yyyy-MM-dd") //将返回的时间结果进行格式化
.missing("2016-05-01"); //缺省字段：当遇到没有值的字段，就会按照该字段的值来计算
//查询数据进行聚合分析
SearchResponse response = EsClient.client()
.prepareSearch("tvs").setTypes("sales")
.addAggregation(dateHistogramInterval)
.setSize(0)
.execute()
.actionGet();
//获得我们设置的聚合
Histogram agg = response.getAggregations().get("date_hist");
//循环数据输出
for (Histogram.Bucket entry : agg.getBuckets()) {
Map<String,Object> map = new HashMap<String,Object>();
//处理
DateTime key = (DateTime) entry.getKey(); // Key
String keyAsString = entry.getKeyAsString(); // Key as String
long docCount = entry.getDocCount(); // Doc count
//封装
map.put("date",keyAsString);
map.put("doc_count",docCount);
System.out.println(map);
}
}这个功能介绍中，我会减少代码量，只把关键代码贴出来，代码中会有注释
3.1.数字范围：
AggregationBuilder rangAgg = AggregationBuilders
.range("rang_price")
.field("price")
.addUnboundedTo(1000) //从负无穷到1000
.addRange(1001, 1500) //从1001-1500
.addRange(1501, 2000) //从1501-2000
.addUnboundedFrom(2001);//2000到正无穷// 获得我们设置的聚合
Range agg = response.getAggregations().get("rang_price");
// 循环数据输出
for (Range.Bucket entry : agg.getBuckets()) {
Map<String, Object> map = new HashMap<String, Object>();
// 处理
String key = entry.getKeyAsString(); // Range as key
Number from = (Number) entry.getFrom(); // Bucket from
Number to = (Number) entry.getTo(); // Bucket to
long docCount = entry.getDocCount(); // Doc count
// 封装
map.put("key", key);
map.put("doc_count", docCount);
System.out.println(map);
}AggregationBuilder aggregation =
AggregationBuilders
.dateRange("agg")
.field("dateOfBirth") //field名称
.format("yyyy") //日期格式
.addUnboundedTo("1950") // from -infinity to 1950 (excluded)
.addRange("1950", "1960") // from 1950 to 1960 (excluded)
.addUnboundedFrom("1960"); // from 1960 to +infinityfor (Range.Bucket entry : agg.getBuckets()) {
String key = entry.getKeyAsString(); // Date range as key
DateTime fromAsDate = (DateTime) entry.getFrom(); // Date bucket from as a Date
DateTime toAsDate = (DateTime) entry.getTo(); // Date bucket to as a Date
long docCount = entry.getDocCount(); // Doc count
}3.3 IP Range Aggregation ip范围分组
IpRangeAggregationBuilder aggregation1 =
AggregationBuilders
.ipRange("agg")
.field("ip")
.addUnboundedTo("192.168.1.0") // 从负无穷到192.168.1.0
.addRange("192.168.1.0", "192.168.2.0") // 从 192.168.1.0 到 192.168.2.0 (excluded)
.addUnboundedFrom("192.168.2.0"); // 从 192.168.2.0 到 +infinityfor (Range.Bucket entry : aggregation1.getBuckets()) {
String key = entry.getKeyAsString(); // Ip 范围 as key
String fromAsString = entry.getFromAsString(); // Ip bucket from as a String
String toAsString = entry.getToAsString(); // Ip bucket to as a String
long docCount = entry.getDocCount(); // Doc count
}AggregationBuilder aggregation =
AggregationBuilders
.filters("agg",
new FiltersAggregator.KeyedFilter("men", QueryBuilders.termQuery("gender", "male")),
new FiltersAggregator.KeyedFilter("women", QueryBuilders.termQuery("gender", "female")));// For each entry
for (Filters.Bucket entry : agg.getBuckets()) {
String key = entry.getKeyAsString(); // bucket key
long docCount = entry.getDocCount(); // Doc count
logger.info("key [{}], doc_count [{}]", key, docCount);
}key [men], doc_count [4982]
key [women], doc_count [5018]三、Metrics aggregations 聚合分析的使用
官方参考文档：
https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/_metrics_aggregations.html
3.1常用的：Avg，Max，Sum等等
我通过一个需求将metric的javaAPI进行说明，需求是：查询电视销售每个颜色的平均价格、最高价格、总价格并按照平均价格降序排序，对应的sql语句为:
select avg(price) avg_price,max(price) max_price,sum(price) sum_price from tvs.sales group by color order by avg_price
/**
* 查询电视销售每个颜色的平均价格、最高价格、总价格并按照平均价格降序排序
* @Title: metrics
* @Description: TODO(这里用一句话描述这个方法的作用)
* @param @throws UnknownHostException
* @return void
* @autor lpl
* @date 2017年11月6日
* @throws
*/
@Test
public void metrics() throws UnknownHostException{
// 创建查询索引和type
SearchRequestBuilder srBuilder =EsClient.client()
.prepareSearch("tvs")
.setTypes("sales");
// colorAggs
// 如果需要进行排序的话，可以使用order(Order.aggregation("avg_price",true))
// "avg_price"对应下班的字聚合的名称，true表示升序排序。false表示倒叙排序
TermsAggregationBuilder colorsAgg = AggregationBuilders
.terms("colors")
.field("color")
.order(Order.aggregation("avg_price", true));
// 平均值
AvgAggregationBuilder avgAgg = AggregationBuilders.avg("avg_price").field("price");
// 最大值a
MaxAggregationBuilder maxAgg = AggregationBuilders.max("max_price").field("price");
// 总和
SumAggregationBuilder sumAgg = AggregationBuilders.sum("sum_price").field("price");
SearchResponse response = srBuilder.setSize(0)
.addAggregation(colorsAgg
.subAggregation(avgAgg)
.subAggregation(maxAgg)
.subAggregation(sumAgg))
.execute()
.actionGet();
Terms terms = response.getAggregations().get("colors");
//处理结果
for (Terms.Bucket entry : terms.getBuckets()) {
Map<String, Object> map = new HashMap<String, Object>();
// 获得按照颜色进行分组的桶
String keyAsString = entry.getKeyAsString();
// 获得每个分组的数量
long docCount = entry.getDocCount();
// 获得平均价格
Avg avg = entry.getAggregations().get("avg_price");
// 获得最大价格
Max max = entry.getAggregations().get("max_price");
// 获得价格总和
Sum sum = entry.getAggregations().get("sum_price");
map.put("key", keyAsString);
map.put("docCount", docCount);
map.put("avg", avg.getValue());
map.put("max", max.getValue());
map.put("sum", sum.getValue());
System.out.println(map);
}
}PercentilesAggregationBuilder aggregation =
AggregationBuilders
.percentiles("agg")
.field("height")
.percentiles(1.0, 5.0, 10.0, 20.0, 30.0, 75.0, 95.0, 99.0);// sr is here your SearchResponse object
Percentiles agg = response.getAggregations().get("agg");
// For each entry
for (Percentile entry : agg) {
double percent = entry.getPercent(); // Percent
double value = entry.getValue(); // Value
}percent [1.0], value [0.814338896154595]
percent [5.0], value [0.8761912455821302]
percent [25.0], value [1.173346540141847]
percent [50.0], value [1.5432023318692198]
percent [75.0], value [1.923915462033674]
percent [95.0], value [2.2273644908535335]
percent [99.0], value [2.284989339108279]
上边就是我对elasticsearch的聚合分析的一些javaAPI使用的总结，当然我只写出了一部分，强大的elasticsearch还有很多的接口，大家可以通过上边我提供的连接进入elasticsearch的官网进行查询。在前面的文章中我也对elasticsearch的常用搜索、curd、高亮显示、搜索建议进行了记录，如果大家有兴趣可以去翻阅。文章中有哪些地方有误请大家及时指出，感激不尽，相互交流，共同进步！
北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台
更多推荐
1
0- 0
已为社区贡献4条内容
所有评论(0)