文档章节

ElasticSearch java API - 聚合查询-聚合多字段聚合demo

xiaomin0322
 xiaomin0322
发布于 07/12 16:58
字数 1242
阅读 433
收藏 7

以球员信息为例,player索引的player type包含5个字段,姓名,年龄,薪水,球队,场上位置。
index的mapping为:

"mappings": {
	"player": {
		"properties": {
			"name": {
				"index": "not_analyzed",
				"type": "string"
			},
			"age": {
				"type": "integer"
			},
			"salary": {
				"type": "integer"
			},
			"team": {
				"index": "not_analyzed",
				"type": "string"
			},
			"position": {
				"index": "not_analyzed",
				"type": "string"
			}
		},
		"_all": {
			"enabled": false
		}
	}
}


索引中的全部数据:
 

微信截图_20160920171030.png


 
首先,初始化Builder:

SearchRequestBuilder sbuilder = client.prepareSearch("player").setTypes("player");

接下来举例说明各种聚合操作的实现方法,因为在es的api中,多字段上的聚合操作需要用到子聚合(subAggregation),初学者可能找不到方法(网上资料比较少,笔者在这个问题上折腾了两天,最后度了源码才彻底搞清楚T_T),后边会特意说明多字段聚合的实现方法。另外,聚合后的排序也会单独说明。

  • group by/count

例如要计算每个球队的球员数,如果使用SQL语句,应表达如下:

select team, count(*) as player_count from player group by team;

ES的java api:

TermsBuilder teamAgg= AggregationBuilders.terms("player_count ").field("team");
sbuilder.addAggregation(teamAgg);
SearchResponse response = sbuilder.execute().actionGet();

 

  • group by多个field

例如要计算每个球队每个位置的球员数,如果使用SQL语句,应表达如下:

select team, position, count(*) as pos_count from player group by team, position;

ES的java api:

TermsBuilder teamAgg= AggregationBuilders.terms("player_count ").field("team");
TermsBuilder posAgg= AggregationBuilders.terms("pos_count").field("position");
sbuilder.addAggregation(teamAgg.subAggregation(posAgg));
SearchResponse response = sbuilder.execute().actionGet();

 

  • max/min/sum/avg

例如要计算每个球队年龄最大/最小/总/平均的球员年龄,如果使用SQL语句,应表达如下:

select team, max(age) as max_age from player group by team;

ES的java api:

TermsBuilder teamAgg= AggregationBuilders.terms("player_count ").field("team");
MaxBuilder ageAgg= AggregationBuilders.max("max_age").field("age");
sbuilder.addAggregation(teamAgg.subAggregation(ageAgg));
SearchResponse response = sbuilder.execute().actionGet();

 

  • 对多个field求max/min/sum/avg

例如要计算每个球队球员的平均年龄,同时又要计算总年薪,如果使用SQL语句,应表达如下:

select team, avg(age)as avg_age, sum(salary) as total_salary from player group by team;

ES的java api:

TermsBuilder teamAgg= AggregationBuilders.terms("team");
AvgBuilder ageAgg= AggregationBuilders.avg("avg_age").field("age");
SumBuilder salaryAgg= AggregationBuilders.avg("total_salary ").field("salary");
sbuilder.addAggregation(teamAgg.subAggregation(ageAgg).subAggregation(salaryAgg));
SearchResponse response = sbuilder.execute().actionGet();

 

  • 聚合后对Aggregation结果排序

例如要计算每个球队总年薪,并按照总年薪倒序排列,如果使用SQL语句,应表达如下:

select team, sum(salary) as total_salary from player group by team order by total_salary desc;

ES的java api:

TermsBuilder teamAgg= AggregationBuilders.terms("team").order(Order.aggregation("total_salary ", false);
SumBuilder salaryAgg= AggregationBuilders.avg("total_salary ").field("salary");
sbuilder.addAggregation(teamAgg.subAggregation(salaryAgg));
SearchResponse response = sbuilder.execute().actionGet();

需要特别注意的是,排序是在TermAggregation处执行的,Order.aggregation函数的第一个参数是aggregation的名字,第二个参数是boolean型,true表示正序,false表示倒序。 

  • Aggregation结果条数的问题

默认情况下,search执行后,仅返回10条聚合结果,如果想反悔更多的结果,需要在构建TermsBuilder 时指定size:

TermsBuilder teamAgg= AggregationBuilders.terms("team").size(15);

 

  • Aggregation结果的解析/输出

得到response后:

Map<String, Aggregation> aggMap = response.getAggregations().asMap();
StringTerms teamAgg= (StringTerms) aggMap.get("keywordAgg");
Iterator<Bucket> teamBucketIt = teamAgg.getBuckets().iterator();
while (teamBucketIt .hasNext()) {
Bucket buck = teamBucketIt .next();
//球队名
String team = buck.getKey();
//记录数
long count = buck.getDocCount();
//得到所有子聚合
Map subaggmap = buck.getAggregations().asMap();
//avg值获取方法
double avg_age= ((InternalAvg) subaggmap.get("avg_age")).getValue();
//sum值获取方法
double total_salary = ((InternalSum) subaggmap.get("total_salary")).getValue();
//...
//max/min以此类推
}

 

  • 总结

综上,聚合操作主要是调用了SearchRequestBuilder的addAggregation方法,通常是传入一个TermsBuilder,子聚合调用TermsBuilder的subAggregation方法,可以添加的子聚合有TermsBuilder、SumBuilder、AvgBuilder、MaxBuilder、MinBuilder等常见的聚合操作。
 
从实现上来讲,SearchRequestBuilder在内部保持了一个私有的 SearchSourceBuilder实例, SearchSourceBuilder内部包含一个List<AbstractAggregationBuilder>,每次调用addAggregation时会调用 SearchSourceBuilder实例,添加一个AggregationBuilder。
同样的,TermsBuilder也在内部保持了一个List<AbstractAggregationBuilder>,调用addAggregation方法(来自父类addAggregation)时会添加一个AggregationBuilder。有兴趣的读者也可以阅读源码的实现。
 
如果有什么问题,欢迎一起讨论,如果文中有什么错误,欢迎批评指正。
 
注:文中使用的Elastic Search API版本为2.3.2

 

 

 public List<Map<String, Object>> queryAggregationsByAttr(BoolQueryBuilder boolQueryBld){
    	 List<Map<String, Object>> result = new ArrayList<>();
    	 
    	 
        
        NestedBuilder nestedBuilder= AggregationBuilders.nested("negstedAttr").path("spuAttrList");  
        //属性名称分组
        TermsBuilder tbName=  AggregationBuilders.terms("attrNameAgg").field("spuAttrList.name");  
        
        //嵌套查询的子查询中分组count  
        TermsBuilder tb=  AggregationBuilders.terms("attrvIdAgg").field("spuAttrList.attrvId");  
        //属性值字段
        TermsBuilder tbVal=  AggregationBuilders.terms("attrValAgg").field("spuAttrList.value");  
        
        
        NestedBuilder all = nestedBuilder.subAggregation(tbName.subAggregation(tb.subAggregation(tbVal)));
        
        
        NativeSearchQueryBuilder nativeQueryBuilderAgg = new NativeSearchQueryBuilder()
        		 .withQuery(boolQueryBld)
        		 .withIndices("skus").withTypes("skus")
        		 .addAggregation(all);
        	        
        SearchQuery searchQueryAgg = nativeQueryBuilderAgg.build();
        
        
        Aggregations aggregations = elasticsearchTemplate.query(searchQueryAgg, new ResultsExtractor<Aggregations>() {  
            @Override  
            public Aggregations extract(SearchResponse response) {  
                return response.getAggregations();  
            }  
        }); 
        
        Map<String, Aggregation> map=aggregations.asMap();  
      
        for(String s:map.keySet()){  
       	 if("negstedAttr".equals(s)) {
       		 InternalNested internalNested  = (InternalNested)map.get(s);
       		 //属性名称
       		 StringTerms nameTerms=(StringTerms) internalNested.getAggregations().get("attrNameAgg");
       		 
       	
       		 //属性子表id
       		 for(org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket tbket:nameTerms.getBuckets()){
       			 
       			         //对应一组属性值
		       			 Map<String, Object> categoryIdsMapTerms = new HashMap<String, Object>();
		       			 categoryIdsMapTerms.put("typeId", "attrValueIds");
		           	     categoryIdsMapTerms.put("typeName", tbket.getKeyAsString());
           	    
       					 LongTerms attrvIdTerms=(LongTerms)tbket.getAggregations().asMap().get("attrvIdAgg");
    		   			 if(attrvIdTerms == null || CollectionUtils.isEmpty(attrvIdTerms.getBuckets())) {
    		             	 continue;
    		              }
    		   			 
    		   			List<Map<String, Object>> dataList = new ArrayList<>();
    		   			
    		   		    //属性子表val
    		   			for(org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket attrIdB : attrvIdTerms.getBuckets()) {
	    		   			 //dataListMap
	       		   			 Map<String, Object> dataListMap = new HashMap<String, Object>();
       		   			 
       		   			 
    		   				 Long attrvId = (Long) attrIdB.getKeyAsNumber();
    		   				    
    		   				 StringTerms valTerms=(StringTerms)  attrIdB.getAggregations().asMap().get("attrValAgg");
    		   				 if(valTerms == null || CollectionUtils.isEmpty(valTerms.getBuckets())) {
        		             	 continue;
        		              }
	    		   			String attrValStr = valTerms.getBuckets().get(0).getKeyAsString();
	    		   			dataListMap.put("id", attrvId);
	    		   			dataListMap.put("name", attrValStr);
	    		   			dataList.add(dataListMap);
	    		   			
    		   			}
    		   	        if(!CollectionUtils.isEmpty(dataList)) {
    		   	        	categoryIdsMapTerms.put("dataList", dataList);
    		   		     }
    		   	         result.add(categoryIdsMapTerms);
       		 }
       	 }
       }  
        
        return result;
    	
    }

 

本文转载自:https://elasticsearch.cn/article/102

共有 人打赏支持
xiaomin0322
粉丝 115
博文 3562
码字总数 166184
作品 0
上海
架构师
私信 提问
ES(elasticsearch)搜索引擎

ES(elasticsearch)搜索引擎 0、授人以渔,少走半年弯路! 死磕 Elasticsearch 方法论:普通程序员高效精进的 10 大狠招! 一、Elasitcsearch基础篇 1.1 Elasitcsearch基础认知 1、Elasticse...

Ocean_K
09/11
0
0
如何优化100s的Elasticsearch 查询到1s以内

在SQL的世界里, 查询优化是相当成熟并且是可以理解的, 另外一方面, 分布式数据库系统是新出现的, 并且不太成熟. 理解查询是如何工作的将是一件非常重要的事情. Elstaticsearch的查询有时候将...

leon_lu
2017/10/31
0
0
(一)ElasticSearch-入门

目录: 一.前言 二.安装 三.索引 四.搜索 五.聚合 六.分布式的特性 一.前言 Elasticsearch是一个基于Apache Lucene(TM)的开源搜索引擎。无论在开源还是专有领域,Lucene可以被认为是迄...

Deng_Wu
08/16
0
0
阿里云Elasticsearch性能优化实践

Elasticsearch是一款流行的分布式开源搜索和数据分析引擎,具备高性能、易扩展、容错性强等特点。它强化了Apache Lucene的搜索能力,把掌控海量数据索引和查询的方式提升到一个新的层次。本文...

小扑
11/16
0
0
Elasticsearch初探(1)——基本介绍与环境搭建

版权声明:本文版权归Jitwxs所有,欢迎转载,但未经作者同意必须保留原文链接。 https://blog.csdn.net/yuanlaijike/article/details/82966110 一、Elasticsearch简介 1.1 什么是Elasticsear...

Jitwxs
10/08
0
0

没有更多内容

加载失败,请刷新页面

加载更多

CAS和AQS

CAS(Compare And Swap) 什么是CAS CAS(Compare And Swap),即比较并交换。是解决多线程并行情况下使用锁造成性能损耗的一种机制。 CAS是一种系统原语,原语属于操作系统用语范畴,是由若干条...

狼王黄师傅
31分钟前
2
0
Python利用pandas处理Excel数据的应用

Python利用pandas处理Excel数据的应用 最近迷上了高效处理数据的pandas,其实这个是用来做数据分析的,如果你是做大数据分析和测试的,那么这个是非常的有用的!!但是其实我们平时在做自动化...

上官夏洛特
45分钟前
2
0
100:代码管理平台svn

1:代码管理平台:代码版本管理工具,工作中用于代码更新,代码发布统计,代码版本控制;(频繁更新代码的版本记录),方便查阅历史变更,协同开发及合并; 版本管理工具发展: cvs(编码兼容不太好)...

芬野de博客
45分钟前
1
0
Linux学习-1101(rsync下、log、screen)

10.32/10.33 rsync通过服务同步 10.34 linux系统日志 10.35 screen工具 一、 rsync通过服务同步 rsync 通过服务的方式同步,首先需要开启一个服务。它是C/S架构,开启rsync服务,默认端口是8...

wxy丶
52分钟前
1
0
使用jpmml-sparkml-executable生成PMML模型文件

加载依赖的jar包通过./spark-shell --jars ./jpmml-sparkml-executable-1.2.13.jar 启动spark import org.apache.spark.ml.classification.LogisticRegressionimport org.apache.spark.m......

KYO4321
55分钟前
1
1

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部