maven依赖

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

配置文件

spring:
  data:       
    elasticsearch:
      client:
        reactive:
          endpoints:
          - es:9200
        

DSL实现

Java实现

    @Autowired
    ReactiveElasticsearchTemplate reactiveElasticsearchTemplate;

    public Flux<org.elasticsearch.search.aggregations.Aggregation> getDrivingTrendModel(Search search) {
        NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
        // 构建布尔查询
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        //精确查询
        boolQueryBuilder.must(QueryBuilders.termQuery("tenantId", search.getTenantId()));
        //过滤时间范围
        boolQueryBuilder.filter(QueryBuilders.rangeQuery("ts").lte(Instant.now()));
        nativeSearchQueryBuilder.withQuery(boolQueryBuilder);
        // 构建聚合查询（dateHistogram 按照日期分组）
        DateHistogramAggregationBuilder dateHistogramAggregationBuilder = AggregationBuilders.dateHistogram("actions_per_day").field("ts").calendarInterval(DateHistogramInterval.DAY).format("yyyy-MM-dd");
        //再求基数（其实相当于该字段互不相同的值有多少类，输出的是种类数，相当于mysql去重计数）
        CardinalityAggregationBuilder cardinalityAggregationBuilder = AggregationBuilders.cardinality("group_by_user").field("username.keyword");
        //是一个子聚合
        dateHistogramAggregationBuilder.subAggregation(cardinalityAggregationBuilder);
        nativeSearchQueryBuilder.addAggregation(dateHistogramAggregationBuilder);
        return reactiveElasticsearchTemplate.aggregate(nativeSearchQueryBuilder.build(), org.elasticsearch.search.aggregations.Aggregation.class, IndexCoordinates.of("taurus-log.tauaction"));
    }