原创ElasticSearch使用Java代码group by多个字段查询统计数量

Posted 2021-10-19 DCTANT

这篇文章主要介绍group by多个字段查询，这方面的资料在全网都非常少，而我这边的需求需要group by三个字段，而不是仅仅一个字段，大大增加了检索资料的难度，还好这问题被我解决了，多亏了公司里的老程序员。 

首先自然是在SpringBoot代码中引入ES查询的client

Maven依赖：

        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.14.0</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>7.14.0</version>
        </dependency>

    @Autowired
    private RestHighLevelClient client;

构建查询请求，并创建查询的Builder，其中的esIndex是个字符串类型，就是ES的索引，然后创建布尔类型查询条件，设置systemId大于0

        SearchRequest searchRequest = new SearchRequest(esIndex);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

重点来了，我这边需要groupBy3个字段，核心是tableName，另外两个是dataSouceId、systemId，如果按照SQL语句来写的话就是

select count(0) as count from esIndex （where条件暂时用不着） group by tableName, dataSourceId, systemId

这里使用Aggregation（中文意思：聚合）关键字进行聚合分组，然后是最关键的，多个group by不能使用并列Aggregation，必须使用嵌套聚合，采用subAggregation进行嵌套！！

        AggregationBuilder aggregationBuilder = AggregationBuilders.terms("tableName").field("tableName").size(100000);
        TermsAggregationBuilder dataSourceAgg = AggregationBuilders.terms("dataSourceId").field("dataSourceId");
        TermsAggregationBuilder systemIdAgg = AggregationBuilders.terms("systemId").field("systemId");
        dataSourceAgg.subAggregation(systemIdAgg);
        aggregationBuilder.subAggregation(dataSourceAgg);

        sourceBuilder.aggregation(aggregationBuilder);

这里的是tableName聚合套dataSourceId聚合，dataSourceId聚合套systemId聚合，总共三层嵌套。

然后进行搜索

        searchRequest.source(sourceBuilder);
        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        Aggregations aggregations = searchResponse.getAggregations();

然后你会发现

searchResponse.getAggregations();搜出来的值也是一个嵌套，即aggregations嵌套buckets，buckets再嵌套aggregations，aggregations再嵌套buckets，等等以此类推。到最底层的buckets.size()才是要获取的count值，需要把这些最底层的buckets.size()累加起来才是最终需要的count值，够坑爹的不是吗。

像我这边3个聚合字段，就需要2*3-1次for循环才能统计成功（最后一层的aggregations的size是0，不需要for循环统计），而且每层结构都一样，那么就可以采用递归统计了。

递归统计代码如下：

    /**
     * 递归统计ES聚合的数量
     * @param aggregations
     * @return
     */
    public static long count(Aggregations aggregations) {
        long totalCount = 0L;
        for (Aggregation aggregation : aggregations) {
            Terms terms = (Terms) aggregation;
            List<? extends Terms.Bucket> buckets = terms.getBuckets();
            if (buckets.size() > 0) {
                if (buckets.get(0).getAggregations().iterator().hasNext()) {
                    // 如果内部还有aggregation，就继续往下走，不能统计
                    for (Terms.Bucket bucket : buckets) {
                        Aggregations aggregationsInners = bucket.getAggregations();
                        if (aggregationsInners == null || aggregationsInners.asList().size() == 0) {
                            System.out.println("进入了aggregationsInners == null || aggregationsInners.asList().size() == 0");
                        } else {
                            totalCount += count(aggregationsInners);
                        }
                    }
                } else {
                    // 到底了，后面没有新的aggregation，可以统计数据了
                    totalCount += buckets.size();
                }
            }
        }
        return totalCount;
    }

 最后得到的count就是count(0)的值。

完整代码如下：（删除了所有BoolQueryBuilder）

        SearchRequest searchRequest = new SearchRequest(esIndex);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        AggregationBuilder aggregationBuilder = AggregationBuilders.terms("tableName").field("tableName").size(100000);
        TermsAggregationBuilder dataSourceAgg = AggregationBuilders.terms("dataSourceId").field("dataSourceId");
        TermsAggregationBuilder systemIdAgg = AggregationBuilders.terms("systemId").field("systemId");
        dataSourceAgg.subAggregation(systemIdAgg);
        aggregationBuilder.subAggregation(dataSourceAgg);

        sourceBuilder.aggregation(aggregationBuilder);
        searchRequest.source(sourceBuilder);

        try {
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            long count = 0;
            Aggregations aggregations = searchResponse.getAggregations();
            count = EsAggregationUtil.count(aggregations);

            return count;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return 0L;