码迷，mamicode.com
首页 > 编程语言 > 详细

Elasticsearch JAVA api轻松搞定groupBy聚合

时间：2017-12-14 00:02:15      阅读：806      评论：0      收藏：0      [点我收藏+]

标签：for   mpm   min()   odi   term   mem   用法   mat   field   

 

本文给出如何使用Elasticsearch的Java API做类似SQL的group by聚合。

为了简单起见，只给出一级groupby即group by field1(而不涉及到多级，例如group by field1, field2, ...)；如果你需要多级的groupby，在实现上可能需要拆分的更加细致。

 

即将给出的方法，适用于如下的场景：

场景1：找出分组中的所有桶，例如，select group_name from index_name group by group_name;

场景2：灵活添加一个或者多个聚合函数，例如，select group_name, max(count), avg(count) group by group_name;

 

1、用法

GroupBy类是我们的实现。

1）测试用例

public static void main(String[] args) {
        /*
        *   初始化es客户端
        * */
        ESClient esClient = new ESClient(
                "dqa-cluster",
                "10.93.21.21:9300,10.93.18.34:9300,10.93.18.35:9300,100.90.62.33:9300,100.90.61.14:9300",
                false);

        /*
        *   为了演示, 构造了一个距离查询, 相当于where子句.
        * */
        GeoDistanceRangeQueryBuilder queryBuilder = QueryBuilders.geoDistanceRangeQuery("location")
                .point(39.971424, 116.398251)
                .from("0m")
                .to(String.format("%fm", 500.0))
                .includeLower(true)
                .includeUpper(true)
                .optimizeBbox("memory")
                .geoDistance(GeoDistance.SLOPPY_ARC);
        
        SearchRequestBuilder search = esClient.getClient().prepareSearch("moon").setTypes("bj")
                .setSearchType(SearchType.DFS_QUERY_AND_FETCH)
                .setQuery(queryBuilder);

        /*
        *  GroupBy类就是我们的实现, 初始化的时候传入的参数依次是, search, 桶命名, 分桶字段, 排序asc
        *  select date as date_group from index group by date;
        * */
        GroupBy groupBy = new GroupBy(search, "date_group", "date", true);

        /*
        *   添加各种分组函数
        *   这里我实现了10种, 下面是其中的6种
        * */
        groupBy.addSumAgg("pre_total_fee_sum", "pre_total_fee");
        groupBy.addAvgAgg("pre_total_fee_avg", "pre_total_fee");
        groupBy.addPercentilesAgg("pre_total_fee_percent", "pre_total_fee");
        groupBy.addPercentileRanksAgg("pre_total_fee_percentRank", "pre_total_fee", new double[]{13, 16, 20});
        groupBy.addStatsAgg("pre_total_fee_stats", "pre_total_fee");
        groupBy.addCardinalityAgg("type_card", "type");

        /*
        *   获取groupBy聚合的结果
        *   结果是两级Map, 这里的实现是TreeMap因为要保护桶的排序
        * */
        Map<String, Object> groupbyResponse = groupBy.getGroupbyResponse();
        for (Map.Entry<String, Object> entry : groupbyResponse.entrySet()) {
            String bucketKey = entry.getKey();
            Map<String, String> subAggMap = (Map<String, String>) entry.getValue();
            System.out.println(String.format("%s\t%s\t%s", bucketKey, "pre_total_fee_sum", subAggMap.get("pre_total_fee_sum")));
            System.out.println(String.format("%s\t%s\t%s", bucketKey, "pre_total_fee_avg", subAggMap.get("pre_total_fee_avg")));
            System.out.println(String.format("%s\t%s\t%s", bucketKey, "pre_total_fee_percent", subAggMap.get("pre_total_fee_percent")));
            System.out.println(String.format("%s\t%s\t%s", bucketKey, "pre_total_fee_percentRank", subAggMap.get("pre_total_fee_percentRank")));
            System.out.println(String.format("%s\t%s\t%s", bucketKey, "pre_total_fee_stats", subAggMap.get("pre_total_fee_stats")));
            System.out.println(String.format("%s\t%s\t%s", bucketKey, "type_card", subAggMap.get("type_card")));

        }
    }

 

2）初始化

初始化的时候，相当于构造了这样一个SQL：select date as date_group from index group by date;

传入search对象，相当于where子句

传入分桶命名， 相当于 as date_group

传入分桶字段，相当于date

传入排序，asc=true

 

3）初始化完成后，可以添加各种聚合函数，也就是场景2。

GroupBy类里实现了10种聚合函数

 

4）读取结果

结果的返回是两级Map，为了保护分桶的排序，实现中使用了TreeMap。

这里需要注意的是，有些聚合函数的返回，并不是一个值，而是一组值，如Percentiles、Stats等等，这里我们把这一组值压缩成JSONString了。

 

5）打印输出

我们以日期进行了分桶，同一个分桶中的聚合结果，sum、avg、cardinality都是单个的值。而percentiles、percentileRanks、stats是压缩的jsonstring。

技术分享图片

 

2、实现

 先上代码，然后在后面进行讲解。

public class GroupBy {

    private SearchRequestBuilder search;

    private String termsName;

    private TermsBuilder termsBuilder;

    private List<Map<String, Object>> subAggList = new ArrayList<Map<String, Object>>();

    public GroupBy(SearchRequestBuilder search, String termsName, String fieldName, boolean asc) {
        this.search = search;
        this.termsName = termsName;
        termsBuilder = AggregationBuilders.terms(termsName).field(fieldName).order(Terms.Order.term(asc)).size(0);
    }

    private void addSubAggList(String aggName, MetricsAggregationBuilder aggBuilder) {
        Map<String, Object> subAgg = new HashMap<String, Object>();
        subAgg.put("aggName", aggName);
        subAgg.put("aggBuilder", aggBuilder);
        subAggList.add(subAgg);
    }

    public void addSumAgg(String aggName, String fieldName) {
        SumBuilder builder = AggregationBuilders.sum(aggName).field(fieldName);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

    public boolean bucketSumAgg(Terms.Bucket bucket, String aggName, MetricsAggregationBuilder aggBuilder, Map<String, String> tmpMap) {
        if (aggBuilder instanceof SumBuilder) {
            tmpMap.put(aggName, bucket.getAggregations().get(aggName).getProperty("value").toString());
            return true;
        } else {
            return false;
        }
    }

    public void addCountAgg(String aggName, String fieldName) {
        ValueCountBuilder builder = AggregationBuilders.count(aggName).field(fieldName);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

    public boolean bucketCountAgg(Terms.Bucket bucket, String aggName, MetricsAggregationBuilder aggBuilder, Map<String, String> tmpMap) {
        if (aggBuilder instanceof ValueCountBuilder) {
            tmpMap.put(aggName, bucket.getAggregations().get(aggName).getProperty("value").toString());
            return true;
        } else {
            return false;
        }
    }

    public void addAvgAgg(String aggName, String fieldName) {
        AvgBuilder builder = AggregationBuilders.avg(aggName).field(fieldName);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

    public boolean bucketAvgAgg(Terms.Bucket bucket, String aggName, MetricsAggregationBuilder aggBuilder, Map<String, String> tmpMap) {
        if (aggBuilder instanceof AvgBuilder) {
            tmpMap.put(aggName, bucket.getAggregations().get(aggName).getProperty("value").toString());
            return true;
        } else {
            return false;
        }
    }

    public void addMinAgg(String aggName, String fieldName) {
        MinBuilder builder = AggregationBuilders.min(aggName).field(fieldName);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

    public boolean bucketMinAgg(Terms.Bucket bucket, String aggName, MetricsAggregationBuilder aggBuilder, Map<String, String> tmpMap) {
        if (aggBuilder instanceof MinBuilder) {
            tmpMap.put(aggName, bucket.getAggregations().get(aggName).getProperty("value").toString());
            return true;
        } else {
            return false;
        }
    }

    public void addMaxAgg(String aggName, String fieldName) {
        MaxBuilder builder = AggregationBuilders.max(aggName).field(fieldName);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

    public boolean bucketMaxAgg(Terms.Bucket bucket, String aggName, MetricsAggregationBuilder aggBuilder, Map<String, String> tmpMap) {
        if (aggBuilder instanceof MaxBuilder) {
            tmpMap.put(aggName, bucket.getAggregations().get(aggName).getProperty("value").toString());
            return true;
        } else {
            return false;
        }
    }

    public void addStatsAgg(String aggName, String fieldName) {
        StatsBuilder builder = AggregationBuilders.stats(aggName).field(fieldName);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

    public boolean bucketStatsAgg(Terms.Bucket bucket, String aggName, MetricsAggregationBuilder aggBuilder, Map<String, String> tmpMap) {
        if (aggBuilder instanceof StatsBuilder) {
            Stats stats = bucket.getAggregations().get(aggName);
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("min", stats.getMin());
            jsonObject.put("max", stats.getMax());
            jsonObject.put("sum", stats.getMax());
            jsonObject.put("count", stats.getCount());
            jsonObject.put("avg", stats.getAvg());
            tmpMap.put(aggName, jsonObject.toJSONString());
            return true;
        } else {
            return false;
        }
    }

    public void addExtendedStatsAgg(String aggName, String fieldName) {
        ExtendedStatsBuilder builder = AggregationBuilders.extendedStats(aggName).field(fieldName);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

    public boolean bucketExtendedStatsAgg(Terms.Bucket bucket, String aggName, MetricsAggregationBuilder aggBuilder, Map<String, String> tmpMap) {
        if (aggBuilder instanceof ExtendedStatsBuilder) {
            ExtendedStats extendedStats = bucket.getAggregations().get(aggName);
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("min", extendedStats.getMin());
            jsonObject.put("max", extendedStats.getMax());
            jsonObject.put("sum", extendedStats.getMax());
            jsonObject.put("count", extendedStats.getCount());
            jsonObject.put("avg", extendedStats.getAvg());
            jsonObject.put("stdDeviation", extendedStats.getStdDeviation());
            jsonObject.put("sumOfSquares", extendedStats.getSumOfSquares());
            jsonObject.put("variance", extendedStats.getVariance());
            tmpMap.put(aggName, jsonObject.toJSONString());
            return true;
        } else {
            return false;
        }
    }

    public void addPercentilesAgg(String aggName, String fieldName) {
        PercentilesBuilder builder = AggregationBuilders.percentiles(aggName).field(fieldName);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

    public void addPercentilesAgg(String aggName, String fieldName, double[] percentiles) {
        PercentilesBuilder builder = AggregationBuilders.percentiles(aggName).field(fieldName).percentiles(percentiles);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

    public boolean bucketPercentilesAgg(Terms.Bucket bucket, String aggName, MetricsAggregationBuilder aggBuilder, Map<String, String> tmpMap) {
        if (aggBuilder instanceof PercentilesBuilder) {
            Percentiles percentiles = bucket.getAggregations().get(aggName);
            JSONObject jsonObject = new JSONObject();
            for (Percentile percentile : percentiles) {
                jsonObject.put(String.valueOf(percentile.getPercent()), percentile.getValue());
            }
            tmpMap.put(aggName, jsonObject.toJSONString());
            return true;
        } else {
            return false;
        }
    }

    public void addPercentileRanksAgg(String aggName, String fieldName, double[] percentiles) {
        PercentileRanksBuilder builder = AggregationBuilders.percentileRanks(aggName).field(fieldName).percentiles(percentiles);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

    public boolean bucketPercentileRanksAgg(Terms.Bucket bucket, String aggName, MetricsAggregationBuilder aggBuilder, Map<String, String> tmpMap) {
        if (aggBuilder instanceof PercentileRanksBuilder) {
            PercentileRanks percentileRanks = bucket.getAggregations().get(aggName);
            JSONObject jsonObject = new JSONObject();
            for (Percentile percentile : percentileRanks) {
                jsonObject.put(String.valueOf(percentile.getPercent()), percentile.getValue());
            }
            tmpMap.put(aggName, jsonObject.toJSONString());
            return true;
        } else {
            return false;
        }
    }

    public void addCardinalityAgg(String aggName, String fieldName) {
        CardinalityBuilder builder = AggregationBuilders.cardinality(aggName).field(fieldName);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

    public boolean bucketCardinalityAgg(Terms.Bucket bucket, String aggName, MetricsAggregationBuilder aggBuilder, Map<String, String> tmpMap) {
        if (aggBuilder instanceof CardinalityBuilder) {
            tmpMap.put(aggName, bucket.getAggregations().get(aggName).getProperty("value").toString());
            return true;
        } else {
            return false;
        }
    }

    public List<Terms.Bucket> getTermsBucket() {
        search.addAggregation(termsBuilder);
        Terms termsGroup = search.get().getAggregations().get(termsName);
        return termsGroup.getBuckets();
    }

    public Map<String, Object> getGroupbyResponse() {
        Map<String, Object> aggResponseMap = new TreeMap<String, Object>();
        for (Terms.Bucket bucket : getTermsBucket()) {
            String bucketKeyAsString = bucket.getKeyAsString();
            Map<String, String> tmpMap = new TreeMap<String, String>();
            for (Map<String, Object> subAgg : subAggList) {
                String subAggName = subAgg.get("aggName").toString();
                MetricsAggregationBuilder subAggBuilder = (MetricsAggregationBuilder) subAgg.get("aggBuilder");
                if (bucketAvgAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketMaxAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketMinAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketSumAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketCountAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketCardinalityAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketPercentileRanksAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketPercentilesAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketExtendedStatsAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketStatsAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
            }
            aggResponseMap.put(bucketKeyAsString, tmpMap);
        }
        return aggResponseMap;
    }
}

 

1）构造函数

构造函数中，核心逻辑是termsBuilder = AggregationBuilders.terms(termsName).field(fieldName).order(Terms.Order.term(asc)).size(0);

实例化了termsBuilder也就是分桶。

后面调用add...函数簇添加聚合函数的时候，都是通过termsBuilder.subAggregation(builder)在分桶的基础上添加了子聚合。

最后在获取结果的时候search.addAggregation(termsBuilder);将termsBuilder添加到查询上，进行聚合查询。

 

2）添加聚合函数add...函数簇

以sum函数为例

public void addSumAgg(String aggName, String fieldName) {
        SumBuilder builder = AggregationBuilders.sum(aggName).field(fieldName);
        termsBuilder.subAggregation(builder);
        addSubAggList(aggName, builder);
    }

a）初始化了一个SumBuilder聚合操作，然后作为termsBuilder的子聚合。

b）addSubAggList方法在subAggList属性(subAggList属性是一个List<Map<String, Object>>)上保存了所有添加了的子聚合的名字和builder。这样做是为了在解析结果的时候，知道是哪种type的聚合(instanceof)，以便使用不同的逻辑去解析。

private void addSubAggList(String aggName, MetricsAggregationBuilder aggBuilder) {
        Map<String, Object> subAgg = new HashMap<String, Object>();
        subAgg.put("aggName", aggName);
        subAgg.put("aggBuilder", aggBuilder);
        subAggList.add(subAgg);
    }

 

3）按类型获取结果

还是以sum函数为例

public boolean bucketSumAgg(Terms.Bucket bucket, String aggName, MetricsAggregationBuilder aggBuilder, Map<String, String> tmpMap) {
        if (aggBuilder instanceof SumBuilder) {
            tmpMap.put(aggName, bucket.getAggregations().get(aggName).getProperty("value").toString());
            return true;
        } else {
            return false;
        }
    }

a）这里先判断了aggBuilder是哪种类型的(instanceof)，如果是SumBuilder类型的，就按照sum的结果类型去读取返回结果。

b）sum的返回结果就是一个值，当遇到percentiles这种类型的，返回结果不是一个值，此时为了简单，我将结果压缩成了jsonstring，也相当于一个值，可以自行参看代码。

c）后面依赖return true实现了一个逻辑，一旦命中了类型，就不再继续判断了，提升效率。

d）tmpMap是外部传入的一个全局接收器，用来存储结果。

 

4）解析所有的子聚合结果

public Map<String, Object> getGroupbyResponse() {
        Map<String, Object> aggResponseMap = new TreeMap<String, Object>();
        for (Terms.Bucket bucket : getTermsBucket()) {
            String bucketKeyAsString = bucket.getKeyAsString();
            Map<String, String> tmpMap = new TreeMap<String, String>();
            for (Map<String, Object> subAgg : subAggList) {
                String subAggName = subAgg.get("aggName").toString();
                MetricsAggregationBuilder subAggBuilder = (MetricsAggregationBuilder) subAgg.get("aggBuilder");
                if (bucketAvgAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketMaxAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketMinAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketSumAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketCountAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketCardinalityAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketPercentileRanksAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketPercentilesAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketExtendedStatsAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
                if (bucketStatsAgg(bucket, subAggName, subAggBuilder, tmpMap)) continue;
            }
            aggResponseMap.put(bucketKeyAsString, tmpMap);
        }
        return aggResponseMap;
    }

这里是解析结果的代码。tmpMap定义为全局接收器。

a）通过遍历subAggList存储的所有子聚合函数，获取所有的子聚合结果，并保存成两级TreeMap。

b）对每个迭代，调用所有的bucket...函数簇，这里通过if判断是否命中类型，如果命中了，就通过continue不再继续检查。

c） aggResponseMap使用treeMap是为了保持bucket的有序。

 

3、十种聚合函数

最后列出我们实现的十种聚合函数，你可以根据自己的需求继续添加。

1）返回单个值：sum、avg、min、max、count、cardinality(有误差)

2）percentiles：分位数查询，传入分位数，获取分位数上的值；percentileRanks，分位数排名查询，传入值，返回对应的分位数；互为逆向操作。

3）stats和extendedStats，extended聚合更详细的信息max、min、avg、sum、平方和、标准差等。

 

Elasticsearch JAVA api轻松搞定groupBy聚合

标签：for   mpm   min()   odi   term   mem   用法   mat   field   

原文地址：http://www.cnblogs.com/kangoroo/p/8033955.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！