使用 Java 进行 ElasticSearch 聚合

Question

我想在我的 java 应用程序中进行聚合。

首先，我使用 curl 压缩了 REST 查询。它看起来像：

curl -XGET 'localhost:9200/analysis/_search?pretty' -H 'Content-Type: 

application/json' -d'
{
  "size": 0,
  "query" : {
      "bool": {
          "must": [
              { "term" : { "customer_id" : 5117 } }
          ]
      }
  },
  "aggs": {
    "customer_id": {
      "terms": {
        "field": "customer_id", 
        "order": {
          "contract_sum": "desc"
        }
      },
      "aggs": {
        "contract_sum": {
          "sum": {
            "field": "contract_sum"
          }
        }
      }
    }
  }
}
'

它返回了我预期的结果

之后我在java中创建了一些代码

    Settings settings = Settings.builder().put("cluster.name", elasticProperties.getElasticClusterName()).build();
    log.info("Initializing ElasticSearch client");
    try (TransportClient client = new PreBuiltTransportClient(settings).addTransportAddress(new InetSocketTransportAddress(
            InetAddress.getByName(elasticProperties.getElasticTransportAddress()), elasticProperties.getElasticTransportPort()))) {

        // Base query
        log.info("Preparing query");
        SearchRequestBuilder requestBuilder = client.prepareSearch(elasticProperties.getElasticIndexName())
                .setTypes(elasticProperties.getElasticTypeName())
                .setSize(Top);

        // Add aggregations
        AggregationBuilder aggregation =
                AggregationBuilders
                        .terms("customer_id")
                        .field("customer_id")
                        //.order(Terms.Order.aggregation("customer_id", "contract_sum", false))
                        .subAggregation(
                                AggregationBuilders.sum("total_contract_sum")
                                .field("contract_sum")
                        );
        requestBuilder.addAggregation(aggregation);
        // Get response
        log.info("Executing query");
        SearchResponse response = requestBuilder.get();

        log.info("Query results:");

        Terms contractSums = response.getAggregations().get("customer_id");
        for (Terms.Bucket bucket : contractSums.getBuckets()) {
            log.info("  " + bucket.getKey() + " ");
        }

问题是：

如何获取当前桶项的“contract_sum”聚合值？

当我在 IntelliJ Idea 中使用调试工具时，它似乎可以

请帮我提供代码示例。

Answer 1

我和我的网上朋友找到了解决办法

log.info("Query results:");
Terms contractSums = response.getAggregations().get("customer_id");
for (Terms.Bucket bucket : contractSums.getBuckets()) {
    Sum aggValue = bucket.getAggregations().get("total_contract_sum");
    DecimalFormat formatter = new DecimalFormat("0.00");
    log.info("  " + bucket.getKey() + " " + formatter.format(aggValue.getValue()));
}

Answer 2

https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/_bucket_aggregations.html

public List<StringBuilder> aggregation(String filterField, Object start, Object end, String... fields) throws IOException {
    List<StringBuilder> results = new LinkedList<>();
    SearchRequest searchRequest = new SearchRequest(this.index);
    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    if (!StringUtils.isEmpty(filterField)){
        QueryBuilder queryBuilder = QueryBuilders.rangeQuery(filterField).gte(start).lte(end);
        searchSourceBuilder.query(queryBuilder);
    }
    AggregationBuilder aggregationBuilder = buildAggregation(fields, 0);

    if(aggregationBuilder == null)
        return results;
    searchSourceBuilder.aggregation(aggregationBuilder);

    searchRequest.source(searchSourceBuilder);
    SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);

    Terms terms = searchResponse.getAggregations().get(fields[0]);
    processBuckets(terms, new Stack<>(), results);
    return results;
}

public AggregationBuilder buildAggregation(String[] fields, int i) {
    if(i == fields.length - 1)
        return aggregationBuilder(fields[i]);

    return aggregationBuilder(fields[i]).subAggregation(buildAggregation(fields, i + 1));
}

private AggregationBuilder aggregationBuilder(String term) {
    return AggregationBuilders
            .terms(term)
            .field(term)
            .size(1000)
            .order(BucketOrder.count(false));
}

public void processBuckets(Terms terms, Stack<String> stack, List<StringBuilder> results) {
    if(terms != null) {
        for (Terms.Bucket bucket : terms.getBuckets()) {
            stack.push(bucket.getKeyAsString() + ":" + bucket.getDocCount() + ";");
            if(bucket.getAggregations().asList().isEmpty()) {
                StringBuilder result = new StringBuilder();
                for (String s : stack) {
                    result.append(s);
                }
                results.add(result);
            }

            for (Aggregation aggregation : bucket.getAggregations().asList()) {
                processBuckets((Terms) aggregation, stack, results);
            }
            stack.pop();
        }
    }
}

Answer 3

public Map<String, Collection<Map<String, Object>>> processAggregation(SearchResponse response) {
    Map<String,Collection<Map<String, Object>>> responseMap = new HashMap<>();
    try {
        if(response.getAggregations() != null) {
            for (Aggregation agg : response.getAggregations().asList()) {
                String key = agg.getName();
                Terms terms = response.getAggregations().get(key);
                Collection<Map<String, Object>> objects = new HashSet<>();
                Map<String, Object> result = new HashMap<>();
                for(Terms.Bucket bucket : terms.getBuckets()){
                    String extLabel = (String) bucket.getKey();
                    Object count = bucket.getDocCount();
                    result.put(extLabel,count);
                }
                objects.add(result);
                responseMap.put(key,objects);
            }
        }
        return responseMap;
    } catch (Exception e) {
        log.error("Unable to process search response");
    }
    return null;
}