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

以球員信息爲例，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
		}
	}
}

索引中的所有數據：
 

 
首先，初始化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;
    	
    }