ES 6.1.3 java聚合查询并获取结果

快来打我* 2023-06-19 05:28 112阅读 0赞

有数据,然后按照host\_ip，category 进行group by，再取traffic\_index的最大值:

PUT traffic_infos/doc/1
    { 
      "host_ip":"10.42.28.10",
      "category":10012,
      "date":"2019-12-03T00:05:00",
      "traffic_index": 100
    }
    PUT traffic_infos/doc/2
    { 
      "host_ip":"10.42.28.11",
      "category":10012,
      "date":"2019-12-03T00:05:00",
      "traffic_index": 1000
    }
    PUT traffic_infos/doc/3
    { 
      "host_ip":"10.42.28.10",
      "category":10013,
      "date":"2019-12-03T00:05:00",
      "traffic_index": 500
    }
    PUT traffic_infos/doc/4
    { 
      "host_ip":"10.42.28.11",
      "category":10013,
      "date":"2019-12-03T00:05:00",
      "traffic_index": 1500
    }
    PUT traffic_infos/doc/5
    { 
      "host_ip":"10.42.28.10",
      "category":10012,
      "date":"2019-12-03T00:10:00",
      "traffic_index": 200
    }
    PUT traffic_infos/doc/6
    { 
      "host_ip":"10.42.28.11",
      "category":10012,
      "date":"2019-12-03T00:10:00",
      "traffic_index": 1200
    }
    PUT traffic_infos/doc/7
    { 
      "host_ip":"10.42.28.10",
      "category":10013,
      "date":"2019-12-03T00:10:00",
      "traffic_index": 700
    }
    PUT traffic_infos/doc/8
    { 
      "host_ip":"10.42.28.11",
      "category":10013,
      "date":"2019-12-03T00:10:00",
      "traffic_index": 1700
    }

结果：

{ 
      "took": 4,
      "timed_out": false,
      "_shards": { 
        "total": 5,
        "successful": 5,
        "skipped": 0,
        "failed": 0
      },
      "hits": { 
        "total": 8,
        "max_score": 0,
        "hits": []
      },
      "aggregations": { 
        "group_ip": { 
          "doc_count_error_upper_bound": 0,
          "sum_other_doc_count": 0,
          "buckets": [
            { 
              "key": "10.42.28.10",
              "doc_count": 4,
              "group_category": { 
                "doc_count_error_upper_bound": 0,
                "sum_other_doc_count": 0,
                "buckets": [
                  { 
                    "key": 10012,
                    "doc_count": 2,
                    "max_index": { 
                      "value": 200
                    }
                  },
                  { 
                    "key": 10013,
                    "doc_count": 2,
                    "max_index": { 
                      "value": 700
                    }
                  }
                ]
              }
            },
            { 
              "key": "10.42.28.11",
              "doc_count": 4,
              "group_category": { 
                "doc_count_error_upper_bound": 0,
                "sum_other_doc_count": 0,
                "buckets": [
                  { 
                    "key": 10012,
                    "doc_count": 2,
                    "max_index": { 
                      "value": 1200
                    }
                  },
                  { 
                    "key": 10013,
                    "doc_count": 2,
                    "max_index": { 
                      "value": 1700
                    }
                  }
                ]
              }
            }
          ]
        }
      }
    }

使用java查询:

import java.util.Map;
    
    import org.elasticsearch.action.search.SearchRequest;
    import org.elasticsearch.action.search.SearchResponse;
    import org.elasticsearch.client.transport.TransportClient;
    import org.elasticsearch.index.query.BoolQueryBuilder;
    import org.elasticsearch.index.query.QueryBuilders;
    import org.elasticsearch.search.aggregations.Aggregation;
    import org.elasticsearch.search.aggregations.AggregationBuilder;
    import org.elasticsearch.search.aggregations.AggregationBuilders;
    import org.elasticsearch.search.aggregations.Aggregations;
    import org.elasticsearch.search.aggregations.bucket.terms.Terms;
    import org.elasticsearch.search.aggregations.metrics.max.Max;
    import org.elasticsearch.search.builder.SearchSourceBuilder;
    
    public class EsForBLHighTotalTraffic2
    { 
        public static void main(String[] args)
        {       
            try
            { 
                String startTime = "2019-12-03T00:00:00";
                String endTime = "2019-12-04T00:00:00";
                TransportClient client = EsOperation.getInstance();
                
                SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
                
                
                BoolQueryBuilder searchQB = QueryBuilders.boolQuery();                
                searchQB.must(QueryBuilders.rangeQuery("date").gte(startTime).lt(endTime)
                        .includeLower(true).includeUpper(false));
                searchSourceBuilder.size(0);
                
                AggregationBuilder aggregation = 
                		AggregationBuilders.terms("group_ip").field("host_ip.keyword")
                		.subAggregation(AggregationBuilders.terms("group_category").field("category")
                				.subAggregation(AggregationBuilders.max("max_index").field("traffic_index"))
                		)
                		.size(Integer.MAX_VALUE);
                
                searchSourceBuilder.aggregation(aggregation);
                
                try
                { 
                    SearchRequest searchRequest = new SearchRequest("traffic_infos");
                    searchRequest.types("doc"); 
                    searchRequest.source(searchSourceBuilder);
                    
                    long haha = System.currentTimeMillis();
                    
                    SearchResponse searchResponse = client.search(searchRequest).get();
                    Aggregations aggregations = searchResponse.getAggregations();
                    //用来检测查询结果对不对，返回查询结果
    // for (Aggregation a:aggregations)
    // { 
    // StringTerms stringTerms= (StringTerms)a;
    // System.out.println(stringTerms);
    // }
                    Terms ip_Aggregations = searchResponse.getAggregations().get("group_ip");
                    for (Terms.Bucket ip_buck : ip_Aggregations.getBuckets())
                    { 
                    	String ip = ip_buck.getKeyAsString();
    					Long ip_docCount = ip_buck.getDocCount();
    					System.out.println(ip);
    					System.out.println(ip_docCount);
                    	Map<String, Aggregation> aggregationMap = ip_buck.getAggregations().getAsMap();
                    	Terms categoryAggregation = (Terms) aggregationMap.get("group_category");
                    	for (Terms.Bucket category_buck : categoryAggregation.getBuckets())
                    	{ 
                    		String category = category_buck.getKeyAsString();
        					Long category_docCount = category_buck.getDocCount();
        					System.out.println(category);
        					System.out.println(category_docCount);
        					double maxIndex = ((Max)category_buck.getAggregations().get("max_index")).getValue();
        					System.out.println(maxIndex);
                    	}
                    }
                }
                catch(Exception e)
                { 
                	e.printStackTrace();
                } 
            }
            catch (Exception e)
            { 
                e.printStackTrace();
            }
            EsOperation es = new EsOperation();
            es.closeClient();
      
        }
    }