ES时间分组统计查询-Toy模板网

这篇具有很好参考价值的文章主要介绍了ES时间分组统计查询。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

//创建索引
PUT /test
//索引结构
PUT /test/_mapping
{
  "properties": {
   
    "insertTime": {
      "type": "date"
    },
    "id": {
      "type": "text",
      "fields": {
        "keyword": {
          "type": "keyword",
          "ignore_above": 256
        }
      }
    },
   
    "pmValue": {
      "type": "double"
    }
  }
}
//批量插入数据
PUT /test/_bulk
{"index":{}}
{"id":1000,"pmValue":20.13,"insertTime":"2022-08-21"}
{"index":{}}
{"id":2000,"pmValue":20.13,"insertTime":"2022-08-22"}
{"index":{}}
{"id":3000,"pmValue":20.13,"insertTime":"2022-08-20"}
{"index":{}}
{"id":4000,"pmValue":20.13,"insertTime":"2022-08-22"}
{"index":{}}
{"id":5000,"pmValue":20.13,"insertTime":"2022-08-22"}
{"index":{}}
{"id":6000,"pmValue":20.13,"insertTime":"2022-08-22"}
{"index":{}}
{"id":7000,"pmValue":20.13,"insertTime":"2022-08-22"}
{"index":{}}
{"id":8000,"pmValue":20.13,"insertTime":"2022-08-19"}

按照id、时间分组求pmValue的平均值
GET /test/_search
{
  
  "size": 0,
  "aggs": {
    "ID": {
      "terms": {
        "field": "id.keyword"
      },
      "aggs": {
        "时间": {
          "date_histogram": {
            "field": "insertTime",
            "interval": "day",
            "format": "yyyy-MM-dd",
            "min_doc_count": 0,
            "extended_bounds": {
              "min": "2022-08-14",
              "max": "2022-08-20"
            }
          },
          "aggs": {
            "平均值": {
              "avg": {
                "field": "pmValue"
              }
            }
          }
        }
      }
    }
  }
}

查询结果：
ES时间分组统计查询,ES,elasticsearch,mybatis,java
以上对应的Java代码

@Autowired
	private RestHighLevelClient client;
public Map<Object,List<Map<String,Object>>> searchFlow(String index,List<Map<String,Object>> paramList){
		//搜索请求
		SearchRequest request = new SearchRequest(index);
		//请求条件构建器,这里和mybatis中的自定义查询有点类型
		SearchSourceBuilder builder = new SearchSourceBuilder();
		BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
		BoolQueryBuilder query = QueryBuilders.boolQuery();
		for (Map<String, Object> map : paramList) {
			BoolQueryBuilder query1 = QueryBuilders.boolQuery();
			map.forEach((k, v) -> {
				query1.must(QueryBuilders.termsQuery(k + ".keyword", v));
			});
			query.should(query1);
		}
		boolQueryBuilder.must(query);


//其他查询条件
		boolQueryBuilder.must(QueryBuilders.termsQuery("字段" + ".keyword", "字段值"));
		

		//id分组
		TermsAggregationBuilder linkAgg = AggregationBuilders.terms("linkAgg").field("id.keyword");
		//时间分组
		String extendedBoundsMin = DateUtil.format(DateUtil.offsetDay(new Date(), -6), "yyyy-MM-dd");
		String extendedBoundsMax = DateUtil.format(new Date(), "yyyy-MM-dd");
		DateHistogramAggregationBuilder dateHistogramAggregationBuilder = AggregationBuilders
				.dateHistogram("timeAgg").field("insertTime").dateHistogramInterval(DateHistogramInterval.DAY)
				.format("yyyy-MM-dd").minDocCount(0).extendedBounds(new ExtendedBounds(extendedBoundsMin,extendedBoundsMax));

		//取出速率的平均值
		AvgAggregationBuilder avgAggregationBuilder = AggregationBuilders.avg("outSpeedAvg").field("pmValue");
		dateHistogramAggregationBuilder.subAggregation(avgAggregationBuilder);
		linkAgg.subAggregation(dateHistogramAggregationBuilder);

		builder.aggregation(linkAgg);
		builder.query(boolQueryBuilder);
		builder.size(10000);
		//把查询添加放入请求中
		request.source(builder);
		//建立SearchResponse
		SearchResponse response;
		Map<Object,List<Map<String,Object>>> map = new HashMap<>(16);
		try {

			response = client.search(request, RequestOptions.DEFAULT);
			Aggregations aggregations = response.getAggregations();
			Terms linkResTerms = aggregations.get("linkAgg");
			List<? extends Terms.Bucket> linkIdBuckets = linkResTerms.getBuckets();

			if (linkIdBuckets != null) {
				for (Terms.Bucket bucket:linkIdBuckets){
					List<Map<String,Object>> list = new ArrayList<>();
					Object linkId = bucket.getKey();
					Aggregations aggregations1 = bucket.getAggregations();
					ParsedDateHistogram dateHistogram =  aggregations1.get("timeAgg");
					List<? extends Histogram.Bucket> timeBuckets = dateHistogram.getBuckets();
					if (timeBuckets != null) {
						for (Histogram.Bucket bucket1:timeBuckets) {
							Map<String,Object> valueMap = new HashMap<>(16);
							String time = bucket1.getKeyAsString();
							Aggregations aggregations2 = bucket1.getAggregations();
							Avg outSpeedAvg = aggregations2.get("outSpeedAvg");
							double value = outSpeedAvg.getValueAsString() == "Infinity"?0:Double.valueOf(outSpeedAvg.getValueAsString());
							valueMap.put("time",time);
							valueMap.put("value",value);
							list.add(valueMap);
						}
						map.put(linkId,list);
					}

				}
			}

		} catch (Exception e) {
			   log.error("es查询异常",e);
		}
		return map;

	}