目录

1、需求 

2、es查询

3、java实现

---

1、需求 

查询需求类似下面sql:

select apiId,bmdm,avg(dyhs)  as avg_dyhs,count(cwm='0') as sbcs from tj_sjb

group by apiId,bmdm

2、es查询

POST /xx_doc/_search

{

    "query": {

        "bool": {

            "must": [],

            "must_not": [],

            "should": []

        }

    },

    "from": 1,

    "size": 0,

    "sort": [],

    "aggs": {

        "pre": {

            "terms": {

                "script": "doc['apiId'].value +'_'+ doc['bmdm'].value",

                "size": 1

            },

            "aggs": {

                "avg_dyhs": {

                    "avg": {

                        "field": "dyhs"

                    }

                },

                "count_sbcs": {

                    "filter": {

                        "term": {

                            "type": "0"

                        }

                    },

                    "aggs": {

                        "sbcs": {

                            "value_count": {

                                "field": "cwm"

                            }

                        }

                    }

                }

            }

        }

    },

    "track_total_hits": true

}

结果

{
    "took": 240,
    "timed_out": false,
    "_shards": {
        "total": 4,
        "successful": 4,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": {
            "value": 133771,
            "relation": "eq"
        },
        "max_score": null,
        "hits": [
        ]
    },
    "aggregations": {
        "pre": {
            "meta": {
            },
            "doc_count_error_upper_bound": 5,
            "sum_other_doc_count": 93042,
            "buckets": [
                {
                    "key": "02024080716254956700000101416459_BM800002232",
                    "doc_count": 40729,
                    "count_sbcs": {
                        "meta": {
                        },
                        "doc_count": 0,
                        "sbcs": {
                            "value": 0
                        }
                    },
                    "avg_dyhs": {
                        "value": 8530.576542512707
                    }
                }
            ]
        }
    }
}

3、java实现

private void esTj(Long startHourTime, Long endHourTime, String formatedTime) {
    Script script = new Script("doc['apiId'].value +'_'+ doc['bmdm'].value+'_'+ " +
            "doc['yyId'].value+'_'+ doc['apiYyId'].value");

    // es查询
    List<WdDytj> wdDytjList = new ArrayList<>(Constants.DEFAULT_SIZE);
    SearchRequest searchRequest = new SearchRequest(EsConstant.XX_DOC);
    AggregationBuilder termsBuilder = AggregationBuilders.terms("by_group").script(script).size(10000);

    AvgAggregationBuilder avgAggBuilder = AggregationBuilders.avg("avg_dyhs").field(
            "dyhs");

    AggregationBuilder counSbcsAggBuilder = AggregationBuilders.filters("count_sbcs",
            new FiltersAggregator.KeyedFilter("sbcs",
                    QueryBuilders.boolQuery().mustNot(QueryBuilders.termsQuery("cwm", "0"))));
    termsBuilder.subAggregation(avgAggBuilder).subAggregation(counSbcsAggBuilder);


    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder.aggregation(termsBuilder);
    BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();


    RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("dysj").gte(startHourTime).lt(endHourTime);
    boolQuery.must(rangeQueryBuilder);
    searchSourceBuilder.query(boolQuery).size(0);
    searchRequest.source(searchSourceBuilder);
    // 执行es中的查询方法
    SearchResponse response = null;
    try {
        // 获取es中查询数据
        response = EsDynamicSource.getCurClient().search(searchRequest, RequestOptions.DEFAULT);
        // 取出es中的数据并且进行判断
        if (response != null) {
            Terms agg = response.getAggregations().get("by_group");
            if (agg != null) {
                for (Terms.Bucket bucket : agg.getBuckets()) {
                    WdDytj wdDytj = new WdDytj();
                    ParsedAvg avgDyhs = (ParsedAvg) bucket.getAggregations().asMap().get("avg_dyhs");
                    ParsedFilters countSbcs = (ParsedFilters) bucket.getAggregations().asMap().get("count_sbcs");
                    wdDytj.setDycs(bucket.getDocCount());
                    wdDytj.setDyhs(avgDyhs.getValue());
                    wdDytj.setSbcs(countSbcs.getBucketByKey("sbcs").getDocCount());
                    wdDytj.setSjbs(formatedTime);
                    String key = bucket.getKeyAsString();
                    //apiId_bmdm_yyId_apiYyId
                    String[] keys = key.split("_");
                    String now = String.valueOf(System.currentTimeMillis());
                    wdDytj.setApiId(keys[0]);
                    wdDytj.setApiYyId(keys[3]);
                    wdDytj.setBmdm(keys[1]);
                    wdDytj.setYyId(keys[2]);
   
                }
            }
        }
    } catch (Exception e) {

    }
}