java 通過聚合查詢實現elasticsearch的group by

通過聚合查詢獲取group by 后的數量

/**
     * 獲取key的個數
     *
     * @param key   要group by的字段名
     * @param index 索引名稱
     * @return id的個數
     */
    public static int getKeyCount(String key, String index) {
        int count = 0;
        TransportClient client = null;
        try {
            client = connectionPool.getConnection();
            if (client == null) {
                throw new Exception("沒有獲取到連接!");
            }
            SearchRequestBuilder search = client.prepareSearch(index);
            //cardinality聚合查詢，相當於groupby字段名
            SearchResponse sr = search.addAggregation(AggregationBuilders.cardinality(key + "_count").field(key)).execute().actionGet();
            //從返回數據提取id總數
            Cardinality result = sr.getAggregations().get(key + "_count");
            long value = result.getValue();
            count = (int) value;
        } catch (InterruptedException e) {

        } catch (Exception e) {
            logger.error("getKeyCount錯誤", e);
        } finally {
            connectionPool.releaseConnection(client);
        }
        return count;
    }

獲取group by后的所有key值

/**
     * 獲取所有key
     *
     * @param key   被group by的字段名
     * @param index 索引名稱
     * @return 所有id
     */
    public static List<String> getAllKey(String key, String index) {
        int keyCount = getKeyCount(key, index);
        List<String> strings = new ArrayList<>();
        TransportClient client = null;
        try {
            client = connectionPool.getConnection();
            if (client == null) {
                throw new Exception("沒有獲取到數據庫連接!");
            }
            SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index);
            //使用聚合，實現去重查詢
            SearchResponse searchResponse = searchRequestBuilder.
                    addAggregation(AggregationBuilders.terms("models").field(key).size(keyCount)).execute().actionGet();
            Terms term = searchResponse.getAggregations().get("models");
            List<? extends Terms.Bucket> buckets = term.getBuckets();
            //遍歷結果，提取出id
            for (Terms.Bucket bucket : buckets) {
                String keyAsString = bucket.getKeyAsString();
                strings.add(keyAsString);
            }
            buckets.clear();
        } catch (InterruptedException e) {

        } catch (Exception e) {
            logger.error("getAllKey錯誤", e);
        } finally {
            connectionPool.releaseConnection(client);
        }

        return strings;
    }