|
|
@ -15,7 +15,6 @@ import org.elasticsearch.search.aggregations.metrics.min.InternalMin;
|
|
|
import org.elasticsearch.search.aggregations.metrics.sum.InternalSum;
|
|
|
import org.elasticsearch.search.aggregations.metrics.valuecount.InternalValueCount;
|
|
|
import org.springframework.stereotype.Component;
|
|
|
import org.springframework.util.StringUtils;
|
|
|
|
|
|
import java.util.*;
|
|
|
|
|
|
@ -26,133 +25,329 @@ import java.util.*;
|
|
|
public class AggregationBuildHandler {
|
|
|
|
|
|
/**
|
|
|
* 递归解析 json
|
|
|
* @param resultList
|
|
|
* @param
|
|
|
* 简单聚合
|
|
|
* @param aggType 聚合类型
|
|
|
* @param aggName 聚合名称
|
|
|
* @param fieldName 聚合字段
|
|
|
* @return
|
|
|
*/
|
|
|
private void expainAggregationResult(List<Map<String, Object>> resultList,Iterator<Terms.Bucket> gradeBucket,String aggKey){
|
|
|
System.out.println("aggKey=" + aggKey);
|
|
|
while (gradeBucket.hasNext()) {
|
|
|
Terms.Bucket b = gradeBucket.next();
|
|
|
String subVal = StringUtils.isEmpty(aggKey) ? b.getKey().toString() : (aggKey + ":" + b.getKey());
|
|
|
for(Aggregation aggregation : b.getAggregations().asList()){
|
|
|
System.out.println("bucketNmae = " + b.getKey().toString() + "; aggName=" + aggregation.getName());
|
|
|
subVal += ";"+ aggregation.getName();
|
|
|
if (aggregation instanceof InternalValueCount) {
|
|
|
Map<String, Object> countMap = new HashMap<>();
|
|
|
InternalValueCount valueCount = (InternalValueCount) aggregation;
|
|
|
countMap.put(subVal, valueCount.getValue() );
|
|
|
resultList.add(countMap);
|
|
|
}else if (aggregation instanceof InternalSum) {
|
|
|
Map<String, Object> sumMap = new HashMap<>();
|
|
|
InternalSum valueCount = (InternalSum) aggregation;
|
|
|
sumMap.put(subVal, valueCount.getValue() );
|
|
|
resultList.add(sumMap);
|
|
|
}else if (aggregation instanceof InternalMax) {
|
|
|
Map<String, Object> maxMap = new HashMap<>();
|
|
|
InternalMax valueCount = (InternalMax) aggregation;
|
|
|
maxMap.put(subVal, valueCount.getValue() );
|
|
|
resultList.add(maxMap);
|
|
|
}else if (aggregation instanceof InternalMin) {
|
|
|
Map<String, Object> minMap = new HashMap<>();
|
|
|
InternalMin valueCount = (InternalMin) aggregation;
|
|
|
minMap.put(subVal, valueCount.getValue() );
|
|
|
resultList.add(minMap);
|
|
|
}else if (aggregation instanceof InternalAvg) {
|
|
|
Map<String, Object> avgMap = new HashMap<>();
|
|
|
InternalAvg valueCount = (InternalAvg) aggregation;
|
|
|
avgMap.put(subVal, valueCount.getValue() );
|
|
|
resultList.add(avgMap);
|
|
|
}else {
|
|
|
Terms terms = (Terms)aggregation;
|
|
|
expainAggregationResult(resultList,terms.getBuckets().iterator(),subVal);
|
|
|
public AbstractAggregationBuilder addAggregationBuilder(String aggType,String aggName, String fieldName){
|
|
|
AbstractAggregationBuilder builder = null;
|
|
|
if(aggType.equals("sum")){
|
|
|
builder = AggregationBuilders.sum(aggName).field(fieldName);
|
|
|
}else if(aggType.equals("count")){
|
|
|
builder = AggregationBuilders.count(aggName).field(fieldName);
|
|
|
}else if(aggType.equals("avg")){
|
|
|
builder = AggregationBuilders.avg(aggName).field(fieldName);
|
|
|
}else if(aggType.equals("max")){
|
|
|
builder = AggregationBuilders.max(aggName).field(fieldName);
|
|
|
}else if(aggType.equals("min")){
|
|
|
builder = AggregationBuilders.min(aggName).field(fieldName);
|
|
|
}
|
|
|
return builder;
|
|
|
}
|
|
|
|
|
|
|
|
|
/**
|
|
|
* 分组后，简单聚合
|
|
|
* @param aggType 聚合类型
|
|
|
* @param aggName 聚合名称
|
|
|
* @param fieldName 聚合字段
|
|
|
* @return
|
|
|
*/
|
|
|
public AbstractAggregationBuilder addTermAggregationBuilder(String termName, String termFieldName,String aggType,String aggName, String fieldName){
|
|
|
AbstractAggregationBuilder builder = null;
|
|
|
TermsBuilder termsBuilder = addTermsBuilder(termName,termFieldName);
|
|
|
if(aggType.equals("sum")){
|
|
|
builder = termsBuilder.subAggregation(AggregationBuilders.sum(aggName).field(fieldName));
|
|
|
}else if(aggType.equals("count")){
|
|
|
builder = termsBuilder.subAggregation(AggregationBuilders.count(aggName).field(fieldName));
|
|
|
}else if(aggType.equals("avg")){
|
|
|
builder = termsBuilder.subAggregation(AggregationBuilders.avg(aggName).field(fieldName));
|
|
|
}else if(aggType.equals("max")){
|
|
|
builder = termsBuilder.subAggregation(AggregationBuilders.max(aggName).field(fieldName));
|
|
|
}else if(aggType.equals("min")){
|
|
|
builder = termsBuilder.subAggregation(AggregationBuilders.min(aggName).field(fieldName));
|
|
|
}
|
|
|
return builder;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* 添加分组
|
|
|
* @param termName
|
|
|
* @param fieldName
|
|
|
* @return
|
|
|
*/
|
|
|
public TermsBuilder addTermsBuilder(String termName, String fieldName){
|
|
|
TermsBuilder termsBuilder= AggregationBuilders.terms(termName).field(fieldName);
|
|
|
return termsBuilder;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
*
|
|
|
* @param client
|
|
|
* @param boolQueryBuilder 查询的过滤条件
|
|
|
* @param aggBuilderList 聚合组
|
|
|
* 聚合组中成员是 单个的聚合查询，其中单个的聚合查询可以嵌套子聚合查询 如：
|
|
|
* 单个聚合 查询 ：TermsBuilder firstAgg= AggregationBuilders.terms("player_count ").field("team");
|
|
|
* 带有子聚合的聚合查询 ：
|
|
|
* TermsBuilder secondAgg= AggregationBuilders.terms("pos_count").field("position")
|
|
|
* .subAggregation(
|
|
|
* AggregationBuilders.dateHistogram("by_year").field("dateOfBirth").interval((DateHistogramInterval.YEAR))
|
|
|
* .subAggregation(
|
|
|
* AggregationBuilders.avg("avg_children").field("children")
|
|
|
* )
|
|
|
* );
|
|
|
* @return 聚合结果集
|
|
|
*/
|
|
|
public Map<String, Aggregation> structAggregationQuery(TransportClient client,String index,String type,
|
|
|
BoolQueryBuilder boolQueryBuilder,
|
|
|
LinkedList<AbstractAggregationBuilder> aggBuilderList){
|
|
|
|
|
|
SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index).setTypes(type).setQuery(boolQueryBuilder);
|
|
|
for(AbstractAggregationBuilder aggBuilder : aggBuilderList){
|
|
|
searchRequestBuilder.addAggregation(aggBuilder);
|
|
|
}
|
|
|
SearchResponse response = searchRequestBuilder.execute().actionGet();
|
|
|
Map<String, Aggregation> map = response.getAggregations().getAsMap();
|
|
|
client.close();
|
|
|
return map;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* es bucket 数据结果解析
|
|
|
* @param rowList 行维度列表
|
|
|
* @param cloumnList 列维度列表
|
|
|
* @param map 聚合查询结果
|
|
|
* @return
|
|
|
*/
|
|
|
public List<Map<String, Object>> dataParsing( List<String> rowList, List<String> cloumnList,Map<String, Aggregation> map){
|
|
|
List<Map<String, Object>> resultList = new LinkedList<Map<String, Object>>();
|
|
|
//合计行数据处理
|
|
|
Map<String, Object> sumResultMap = new HashMap<>();
|
|
|
sumResultMap.put("区县_terms","合计");
|
|
|
for(String rowAggCode: rowList) {
|
|
|
if (rowAggCode.contains("合计")) {
|
|
|
for (String cloumnAggCode : cloumnList) {
|
|
|
|
|
|
for(String aggCode:map.keySet()) {
|
|
|
Aggregation aggregation = map.get(aggCode);
|
|
|
if (aggregation.getName().equals("合计_" + cloumnAggCode)) {
|
|
|
Map<String, Object> subDataMap = getInternalAggValue(cloumnAggCode, aggregation);
|
|
|
if (subDataMap == null || subDataMap.size() == 0) {
|
|
|
Terms subTerm = (Terms) aggregation;
|
|
|
Map<String, Object> data = getTermAggregationData(subTerm.getBuckets().iterator(), cloumnAggCode);
|
|
|
sumResultMap.putAll(data);
|
|
|
} else {
|
|
|
sumResultMap.putAll(subDataMap);
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
resultList.add(sumResultMap);
|
|
|
|
|
|
|
|
|
//非合计数据处理
|
|
|
for(String rowAggCode: rowList){
|
|
|
if( !rowAggCode.contains("合计")){
|
|
|
for(String aggCode:map.keySet()){
|
|
|
Aggregation aggregation = map.get(aggCode);
|
|
|
String aggName = aggregation.getName();
|
|
|
if(rowAggCode.equals(aggName)){
|
|
|
Terms terms = (Terms)aggregation;
|
|
|
String termNmae = terms.getName();
|
|
|
Map<String, Object> resultMap = new HashMap<>();
|
|
|
String key = "";
|
|
|
List<Terms.Bucket> buckets = terms.getBuckets();
|
|
|
for(Terms.Bucket bucket : buckets){
|
|
|
key = bucket.getKey().toString();
|
|
|
resultMap.put(termNmae,key);
|
|
|
for(String cloumnAggCode: cloumnList){
|
|
|
List<Aggregation> aggregationList = getAggregationList(terms.getBuckets().iterator());
|
|
|
for(Aggregation subAgg : aggregationList){
|
|
|
if(subAgg.getName().equals(cloumnAggCode)){
|
|
|
Map<String, Object> subDataMap = getInternalAggValue(subAgg.getName(),subAgg);
|
|
|
if(subDataMap == null || subDataMap.size() == 0){
|
|
|
Terms subTerm = (Terms)subAgg;
|
|
|
Map<String, Object> data = getTermAggregationData(subTerm.getBuckets().iterator(), cloumnAggCode);
|
|
|
resultMap.putAll(data);
|
|
|
}else {
|
|
|
resultMap.putAll(subDataMap);
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
resultList.add(resultMap);
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
return resultList;
|
|
|
}
|
|
|
|
|
|
public void parsingAggregation(List<Map<String, Object>> resultList,Aggregation aggregation){
|
|
|
private List<Aggregation> getAggregationList(Iterator<Terms.Bucket> gradeBucket){
|
|
|
List<Aggregation> aggregationList = new ArrayList<>();
|
|
|
while (gradeBucket.hasNext()) {
|
|
|
Terms.Bucket b = gradeBucket.next();
|
|
|
aggregationList = b.getAggregations().asList();
|
|
|
}
|
|
|
return aggregationList;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* 查询未分组-统计值
|
|
|
* @param aggregation
|
|
|
* @return
|
|
|
*/
|
|
|
public Map<String, Object> getInternalAggValue(String key, Aggregation aggregation){
|
|
|
Map<String, Object> map = new HashMap<>();
|
|
|
if (aggregation instanceof InternalValueCount) {
|
|
|
Map<String, Object> countMap = new HashMap<>();
|
|
|
InternalValueCount valueCount = (InternalValueCount) aggregation;
|
|
|
countMap.put(valueCount.getName() + ":"+ aggregation.getName(), valueCount.getValue() );
|
|
|
System.out.println(valueCount.getName());
|
|
|
resultList.add(countMap);
|
|
|
map.put(key, valueCount.getValue());
|
|
|
}else if (aggregation instanceof InternalSum) {
|
|
|
Map<String, Object> sumMap = new HashMap<>();
|
|
|
InternalSum valueCount = (InternalSum) aggregation;
|
|
|
sumMap.put(valueCount.getName(), valueCount.getValue() );
|
|
|
System.out.println(valueCount.getName());
|
|
|
resultList.add(sumMap);
|
|
|
map.put(key, valueCount.getValue());
|
|
|
}else if (aggregation instanceof InternalMax) {
|
|
|
Map<String, Object> maxMap = new HashMap<>();
|
|
|
InternalMax valueCount = (InternalMax) aggregation;
|
|
|
maxMap.put(valueCount.getName(), valueCount.getValue() );
|
|
|
resultList.add(maxMap);
|
|
|
map.put(key, valueCount.getValue());
|
|
|
}else if (aggregation instanceof InternalMin) {
|
|
|
Map<String, Object> minMap = new HashMap<>();
|
|
|
InternalMin valueCount = (InternalMin) aggregation;
|
|
|
minMap.put(valueCount.getName(), valueCount.getValue() );
|
|
|
resultList.add(minMap);
|
|
|
map.put(key, valueCount.getValue());
|
|
|
}else if (aggregation instanceof InternalAvg) {
|
|
|
Map<String, Object> avgMap = new HashMap<>();
|
|
|
InternalAvg valueCount = (InternalAvg) aggregation;
|
|
|
avgMap.put(valueCount.getName(), valueCount.getValue() );
|
|
|
resultList.add(avgMap);
|
|
|
}else {
|
|
|
Terms terms = (Terms)aggregation;
|
|
|
String termName = terms.getName();
|
|
|
System.out.println("termName:" + termName);
|
|
|
expainAggregationResult(resultList, terms.getBuckets().iterator(), termName);
|
|
|
map.put(key, valueCount.getValue());
|
|
|
}
|
|
|
return map;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* 行列数据转换
|
|
|
* @param dataList
|
|
|
* @param rowDimensionList
|
|
|
* @param columnDimensionList
|
|
|
* 查询带分组-统计值
|
|
|
* @param gradeBucket
|
|
|
* @param aggCode
|
|
|
* @return
|
|
|
*/
|
|
|
public List<Map<String, Object>> convertList(List<Map<String, Object>> dataList,LinkedList<String> rowDimensionList ,LinkedList<String> columnDimensionList){
|
|
|
List<Map<String,Object>> dataMapList = new ArrayList<>();
|
|
|
List<String> rowList = new ArrayList<>();
|
|
|
for(String rowDimen : rowDimensionList){
|
|
|
for(Map<String, Object> map : dataList) {
|
|
|
for (String key : map.keySet()) {
|
|
|
List<String> result = Arrays.asList(key.split(";"));//区县_terms:信州区 ;性别_terms ; 合计_count_result
|
|
|
for(String str : result){
|
|
|
if(str.contains(rowDimen)){
|
|
|
if(!rowList.contains(str)){
|
|
|
rowList.add(str);
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
Map<String,Object> dataMap = new HashMap<>();
|
|
|
for(String row : rowList){
|
|
|
dataMap.put(row,row);
|
|
|
for(String columnDimen : columnDimensionList){
|
|
|
|
|
|
for(Map<String, Object> map : dataList) {
|
|
|
for (String key : map.keySet()) {
|
|
|
if(key.contains(row) && key.contains(columnDimen)){
|
|
|
List<String> result = Arrays.asList(key.split(";"));
|
|
|
for(String str : result){
|
|
|
dataMap.put(columnDimen,columnDimen);
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
private Map<String, Object> getTermAggregationData(Iterator<Terms.Bucket> gradeBucket ,String aggCode){
|
|
|
Map<String, Object> map = new HashMap<>();
|
|
|
while (gradeBucket.hasNext()) {
|
|
|
Terms.Bucket b = gradeBucket.next();
|
|
|
String subVal = aggCode+"-" + b.getKey();
|
|
|
for(Aggregation aggregation : b.getAggregations().asList()){
|
|
|
if (aggregation instanceof InternalValueCount) {
|
|
|
InternalValueCount valueCount = (InternalValueCount) aggregation;
|
|
|
map.put(subVal, valueCount.getValue() );
|
|
|
}else if (aggregation instanceof InternalSum) {
|
|
|
InternalSum valueCount = (InternalSum) aggregation;
|
|
|
map.put(subVal, valueCount.getValue() );
|
|
|
}else if (aggregation instanceof InternalMax) {
|
|
|
InternalMax valueCount = (InternalMax) aggregation;
|
|
|
map.put(subVal, valueCount.getValue());
|
|
|
}else if (aggregation instanceof InternalMin) {
|
|
|
InternalMin valueCount = (InternalMin) aggregation;
|
|
|
map.put(subVal, valueCount.getValue());
|
|
|
}else if (aggregation instanceof InternalAvg) {
|
|
|
InternalAvg valueCount = (InternalAvg) aggregation;
|
|
|
map.put(subVal, valueCount.getValue());
|
|
|
}else {
|
|
|
Terms terms = (Terms)aggregation;
|
|
|
getTermAggregationData(terms.getBuckets().iterator(), aggCode);
|
|
|
}
|
|
|
}
|
|
|
dataMapList.add(dataMap);
|
|
|
}
|
|
|
//未调试完成
|
|
|
return dataMapList;
|
|
|
return map;
|
|
|
}
|
|
|
|
|
|
|
|
|
// /**
|
|
|
// * 递归解析 json
|
|
|
// * @param resultList
|
|
|
// * @param
|
|
|
// */
|
|
|
// private void expainAggregationResult(List<Map<String, Object>> resultList,Iterator<Terms.Bucket> gradeBucket,String aggKey){
|
|
|
// System.out.println("aggKey=" + aggKey);
|
|
|
// while (gradeBucket.hasNext()) {
|
|
|
// Terms.Bucket b = gradeBucket.next();
|
|
|
// String subVal = StringUtils.isEmpty(aggKey) ? b.getKey().toString() : (aggKey + ":" + b.getKey());
|
|
|
// for(Aggregation aggregation : b.getAggregations().asList()){
|
|
|
// System.out.println("bucketNmae = " + b.getKey().toString() + "; aggName=" + aggregation.getName());
|
|
|
// subVal += ";"+ aggregation.getName();
|
|
|
// if (aggregation instanceof InternalValueCount) {
|
|
|
// Map<String, Object> countMap = new HashMap<>();
|
|
|
// InternalValueCount valueCount = (InternalValueCount) aggregation;
|
|
|
// countMap.put(subVal, valueCount.getValue() );
|
|
|
// resultList.add(countMap);
|
|
|
// }else if (aggregation instanceof InternalSum) {
|
|
|
// Map<String, Object> sumMap = new HashMap<>();
|
|
|
// InternalSum valueCount = (InternalSum) aggregation;
|
|
|
// sumMap.put(subVal, valueCount.getValue() );
|
|
|
// resultList.add(sumMap);
|
|
|
// }else if (aggregation instanceof InternalMax) {
|
|
|
// Map<String, Object> maxMap = new HashMap<>();
|
|
|
// InternalMax valueCount = (InternalMax) aggregation;
|
|
|
// maxMap.put(subVal, valueCount.getValue() );
|
|
|
// resultList.add(maxMap);
|
|
|
// }else if (aggregation instanceof InternalMin) {
|
|
|
// Map<String, Object> minMap = new HashMap<>();
|
|
|
// InternalMin valueCount = (InternalMin) aggregation;
|
|
|
// minMap.put(subVal, valueCount.getValue() );
|
|
|
// resultList.add(minMap);
|
|
|
// }else if (aggregation instanceof InternalAvg) {
|
|
|
// Map<String, Object> avgMap = new HashMap<>();
|
|
|
// InternalAvg valueCount = (InternalAvg) aggregation;
|
|
|
// avgMap.put(subVal, valueCount.getValue() );
|
|
|
// resultList.add(avgMap);
|
|
|
// }else {
|
|
|
// Terms terms = (Terms)aggregation;
|
|
|
// expainAggregationResult(resultList,terms.getBuckets().iterator(),subVal);
|
|
|
// }
|
|
|
// }
|
|
|
// }
|
|
|
// }
|
|
|
//
|
|
|
// public void parsingAggregation(List<Map<String, Object>> resultList,Aggregation aggregation){
|
|
|
// if (aggregation instanceof InternalValueCount) {
|
|
|
// Map<String, Object> countMap = new HashMap<>();
|
|
|
// InternalValueCount valueCount = (InternalValueCount) aggregation;
|
|
|
// countMap.put(aggregation.getName(), valueCount.getValue() );
|
|
|
// System.out.println(valueCount.getName());
|
|
|
// resultList.add(countMap);
|
|
|
// }else if (aggregation instanceof InternalSum) {
|
|
|
// Map<String, Object> sumMap = new HashMap<>();
|
|
|
// InternalSum valueCount = (InternalSum) aggregation;
|
|
|
// sumMap.put(valueCount.getName(), valueCount.getValue() );
|
|
|
// System.out.println(valueCount.getName());
|
|
|
// resultList.add(sumMap);
|
|
|
// }else if (aggregation instanceof InternalMax) {
|
|
|
// Map<String, Object> maxMap = new HashMap<>();
|
|
|
// InternalMax valueCount = (InternalMax) aggregation;
|
|
|
// maxMap.put(valueCount.getName(), valueCount.getValue() );
|
|
|
// resultList.add(maxMap);
|
|
|
// }else if (aggregation instanceof InternalMin) {
|
|
|
// Map<String, Object> minMap = new HashMap<>();
|
|
|
// InternalMin valueCount = (InternalMin) aggregation;
|
|
|
// minMap.put(valueCount.getName(), valueCount.getValue() );
|
|
|
// resultList.add(minMap);
|
|
|
// }else if (aggregation instanceof InternalAvg) {
|
|
|
// Map<String, Object> avgMap = new HashMap<>();
|
|
|
// InternalAvg valueCount = (InternalAvg) aggregation;
|
|
|
// avgMap.put(valueCount.getName(), valueCount.getValue() );
|
|
|
// resultList.add(avgMap);
|
|
|
// }else {
|
|
|
// Terms terms = (Terms)aggregation;
|
|
|
// String termName = terms.getName();
|
|
|
// System.out.println("termName:" + termName);
|
|
|
// expainAggregationResult(resultList, terms.getBuckets().iterator(), termName);
|
|
|
// }
|
|
|
// }
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
}
|
