es java api 进行聚合+桶聚合查询 terms+top_hits+max

布满荆棘的人生 2023-02-21 11:41 3阅读 0赞

### 假设1个member有多个参加的meeting（会议），每个meeting可多次参加，每次参加对应一条参加时间和备注记录 ###

需求： 获取某个member的所有meeting的最新一条记录

查询语句（产生歧义的地方已修改）

{
        
      "query": {
        
        "term": {
        
          "memberid": "2136476"
        }
      },
      "aggs": {
        
        "group": {
        
          "terms": {
        
            "field": "meetingid",
            "order": {
        
              "sortTime": "desc"
            }
          },
          "aggs": {
        
            "result": {
        
              "top_hits": {
        
                "_source": [
                  "meetingid",
                  "memo",
                  "addtime"
                ],
                "size": 1,
                "sort": {
        
                  "addtime": {
        
                    "order": "desc"
                  }
                }
              }
            },
            "sortTime": {
        
              "max": {
        
                "field": "addtime"
              }
            }
          }
        }
      }
    }

Java实现

public List<Map<String, Object>> getaa() {
        
            // 根据客户id查询
            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
            boolQuery.must(QueryBuilders.termQuery("memberid", 2136476));
            SearchSourceBuilder query = new SearchSourceBuilder();
            // 基本查询
            query.query(boolQuery);
    
            // 排序规则
            AggregationBuilder maxTime = AggregationBuilders.max("addtime").field("addtime");
            // 获取每个分组时间倒序排列的第一条记录
            AggregationBuilder top = AggregationBuilders.topHits("result").fetchSource(new String[]{
        "meetingid","memo","addtime"}, null)
                    .size(1).sort("addtime", SortOrder.DESC);
            // 根据会议id进行分组
            TermsAggregationBuilder topBuilder=AggregationBuilders.terms("meeting").field("meetingid");
            // 返回分组数
            topBuilder.size(10000);
            // 分组排序规则
            topBuilder.order(Terms.Order.aggregation("addtime", false));
            topBuilder.subAggregation(top);
            topBuilder.subAggregation(maxTime);
            // 聚合查询
            query.aggregation(topBuilder);
    
            SearchRequest request = Requests.searchRequest("crm").types("meeting_member").
                    searchType(SearchType.QUERY_THEN_FETCH )
                    .source(query);
            SearchResponse response = client.search(request).actionGet();
    
            // 需要返回的list
            List<Map<String, Object>> list = new ArrayList<>();
    
            // 获取查询结果的aggregation部分
            Aggregations aggregations = response.getAggregations();
            Aggregation a = aggregations.get("meeting");
            LongTerms teamSum= (LongTerms)a;
            // 遍历桶聚合后的每一个结果集
            for(Terms.Bucket bucket:teamSum.getBuckets()){
        
                // 获取name为result的top_hits结果集
                TopHits topHits = bucket.getAggregations().get("result");
                SearchHits hits = topHits.getHits();
                SearchHit[] hitArray = hits.getHits();
                // 因为top_hits的siez=1所以不进行遍历直接取第一条数据
                SearchHit hit = hitArray[0];
                Map<String, Object> sourceMap = hit.getSource();
                list.add(sourceMap);
            }
            return list;
        }