[大数据]代码示例------elasticSearch collapse aggr MultiSearch

package com.xxxx.rec.goodsguesslike.impl;

import com.alibaba.dubbo.config.annotation.Service;
import com.gome.boot.common.component.ConfigTool;
import com.gome.rec.goodsguesslike.helper.RestHighLevelClientHelper;
import com.gome.rec.internal.api.es.IGoodsUrlService;
import org.apache.commons.collections.CollectionUtils;
import org.elasticsearch.action.search.MultiSearchRequest;
import org.elasticsearch.action.search.MultiSearchResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.query.TermsQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.collapse.CollapseBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
@Service
public class GoodsUrlServiceImpl implements IGoodsUrlService {
    private Logger logger = LoggerFactory.getLogger(GoodsUrlServiceImpl.class);

    @Autowired
    private RestHighLevelClientHelper restHighLevelClientHelper;
    @Autowired
    private ConfigTool configTool;

    @Override
    // 去重操作 collapse
    public Map<String, String> getFirstGoods(List<String> shopIds, List<String> c1ids, List<String> c2ids, List<String> c3ids) {
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        boolQueryBuilder.filter(QueryBuilders.existsQuery("productImg").boost(0))
                .must(QueryBuilders.termsQuery("shopId", shopIds).boost(0)).boost(0);
        if (CollectionUtils.isNotEmpty(c1ids)) {
            TermsQueryBuilder c1idsQuery = QueryBuilders.termsQuery("cat1", c1ids);
            boolQueryBuilder.should(c1idsQuery).boost(0);
        }
        if (CollectionUtils.isNotEmpty(c2ids)) {
            TermsQueryBuilder c2idsQuery = QueryBuilders.termsQuery("cat2", c2ids);
            boolQueryBuilder.should(c2idsQuery).boost(0);
        }
        if (CollectionUtils.isNotEmpty(c3ids)) {
            TermsQueryBuilder c3idsQuery = QueryBuilders.termsQuery("cat3", c3ids);
            boolQueryBuilder.should(c3idsQuery).boost(0);
        }

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        String[] includes = new String[]{"shopId", "skuId", "productImg"};
        String[] excludes = new String[]{""};
        CollapseBuilder distinctShopId = new CollapseBuilder("shopId");
        searchSourceBuilder.fetchSource(includes, excludes)
                .timeout(TimeValue.timeValueMillis(500))
                .query(boolQueryBuilder)
                .collapse(distinctShopId)
                .sort("scoreAll", SortOrder.DESC);

        String esIndice = configTool.get("es.sku.tableName");
        SearchRequest searchRequest = new SearchRequest(esIndice);
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = restHighLevelClientHelper.search(searchRequest, RequestOptions.DEFAULT, 100, "getFirstGoods:  ");
        if (searchResponse == null) {
            return null;
        }
        SearchHits hits = searchResponse.getHits();
        if (hits == null) {
            return null;
        }
        SearchHit[] searchHits = hits.getHits();
        if (searchHits == null || searchHits.length == 0) {
            return null;
        }
        HashMap<String, String> shopIpToUrl = new HashMap<>();
        for (SearchHit searchHit : searchHits) {
            Map<String, Object> sourceAsMap = searchHit.getSourceAsMap();
            String shopId = (String) sourceAsMap.get("shopId");
            String productImg = (String) sourceAsMap.get("productImg");
            shopIpToUrl.put(shopId, productImg);
        }
        if (shopIpToUrl.isEmpty()) {
            return null;
        }

        return shopIpToUrl;
    }







    //@Override
    // 聚合操作 Aggregate
    public Map<String, String> getFirstGoodsAggregate(List<String> shopIds, List<String> c1ids, List<String> c2ids, List<String> c3ids) {
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        if (CollectionUtils.isNotEmpty(shopIds)) {
            boolQueryBuilder.must(QueryBuilders.termsQuery("shopId", shopIds));
            boolQueryBuilder.filter(QueryBuilders.existsQuery("productImg")).boost(0);
//            boolQueryBuilder.must(QueryBuilders.boolQuery().should(QueryBuilders.termsQuery("shopId",shopIds)).minimumShouldMatch(1));
        }

        if (CollectionUtils.isNotEmpty(c1ids)) {
            TermsQueryBuilder c1idsQuery = QueryBuilders.termsQuery("cat1", c1ids);
            boolQueryBuilder.should(c1idsQuery).boost(0);
        }
        if (CollectionUtils.isNotEmpty(c2ids)) {
            TermsQueryBuilder c2idsQuery = QueryBuilders.termsQuery("cat2", c2ids);
            boolQueryBuilder.should(c2idsQuery).boost(0);
        }
        if (CollectionUtils.isNotEmpty(c3ids)) {
            TermsQueryBuilder c3idsQuery = QueryBuilders.termsQuery("cat3", c3ids);
            boolQueryBuilder.should(c3idsQuery).boost(0);
        }

        String firstBucketName = "groupBYShopId";
        String secondBucketName = "GroupByProductImg";
        AggregationBuilder aggs = AggregationBuilders.terms(firstBucketName).field("shopId");
        TermsAggregationBuilder productImgAggr = AggregationBuilders.terms(secondBucketName).field("productImg").size(1);
        aggs.subAggregation(productImgAggr);
        String esIndice = configTool.get("es.sku.tableName");
        SearchRequest searchRequest = new SearchRequest(esIndice);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder().timeout(TimeValue.timeValueMillis(400)).query(boolQueryBuilder).size(0).aggregation(aggs);
        searchRequest.source(sourceBuilder);

        SearchResponse search = restHighLevelClientHelper
                .search(searchRequest, RequestOptions.DEFAULT, 100, "agger---getFirstGoods:  ");
        Aggregations aggrResults = search.getAggregations();

        Terms firstAggregation = aggrResults.get(firstBucketName);

        List<? extends Terms.Bucket> buckets = firstAggregation.getBuckets();
        if (buckets == null) {
            return null;
        }
        HashMap<String, String> shopIdToUrl = new HashMap<>();
        buckets.forEach(
                bucket -> {
                    String shopId = bucket.getKeyAsString();
                    Terms subAggregations = (Terms) bucket.getAggregations().get(secondBucketName);
                    if (subAggregations == null) {
                        return;
                    }
                    List<? extends Terms.Bucket> urlImgs = subAggregations.getBuckets();
                    if (urlImgs == null || urlImgs.size() == 0) {
                        return;
                    }
                    String productImg = urlImgs.get(0).getKeyAsString();
                    shopIdToUrl.put(shopId, productImg);
                }
        );

        return shopIdToUrl;
    }







    //@Override
    // 批量查询阶段 MultiSearch
    public Map<String, String> getFirstGoodsMultiSearch(List<String> shopIds, List<String> c1ids, List<String> c2ids, List<String> c3ids) {

        if (shopIds == null || shopIds.size() == 0) {
            logger.warn("shopIds is null || size == 0");
            return null;
        }
        /*
        整体流程：
         每个查询主体分两部分拼接： （c1ids，c2ids，c3ids）与 shopId(pop)
         每个查询返回评分最高的一个值
         将本次所有查询使用批量的方式执行
         */
        BoolQueryBuilder subBoolQueryBuilder = null;
        if (CollectionUtils.isNotEmpty(c1ids) || CollectionUtils.isNotEmpty(c2ids) || CollectionUtils.isNotEmpty(c3ids)) {
            subBoolQueryBuilder = new BoolQueryBuilder();
            if (CollectionUtils.isNotEmpty(c1ids)) {
                TermsQueryBuilder c1idsQuery = QueryBuilders.termsQuery("cat1", c1ids);
                subBoolQueryBuilder.should(c1idsQuery).boost(0);
            }
            if (CollectionUtils.isNotEmpty(c2ids)) {
                TermsQueryBuilder c2idsQuery = QueryBuilders.termsQuery("cat2", c2ids);
                subBoolQueryBuilder.should(c2idsQuery).boost(0);
            }
            if (CollectionUtils.isNotEmpty(c3ids)) {
                TermsQueryBuilder c3idsQuery = QueryBuilders.termsQuery("cat3", c3ids);
                subBoolQueryBuilder.should(c3idsQuery).boost(0);
            }
            subBoolQueryBuilder.minimumShouldMatch(1);
        }

        // 批量查询
        MultiSearchRequest multiSearchRequest = new MultiSearchRequest();
        String esIndice = configTool.get("es.sku.tableName");

        for (String shopId : shopIds) {
            SearchRequest searchRequest = new SearchRequest(esIndice);
            TermQueryBuilder shopIdQuery = QueryBuilders.termQuery("shopId", shopId);
            BoolQueryBuilder mainBoolQueryBuilder = new BoolQueryBuilder();
            mainBoolQueryBuilder.must(shopIdQuery);
            if (subBoolQueryBuilder != null) {
                mainBoolQueryBuilder.should(subBoolQueryBuilder);
            }
            mainBoolQueryBuilder.filter(QueryBuilders.existsQuery("productImg")).boost(0);

            // 每个查询返回评分最高的一个值,尝试设置500ms的查询超时限制
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder().from(0).size(1)
                    .timeout(TimeValue.timeValueMillis(500)).query(mainBoolQueryBuilder);
            searchRequest.source(sourceBuilder);
            multiSearchRequest.add(searchRequest);
//
//            // todo
//            String dsl = searchRequest.source().toString();
//            System.out.println("dsl = " + dsl);
        }


        MultiSearchResponse multiSearchResponse = restHighLevelClientHelper.searchBulk(multiSearchRequest, RequestOptions.DEFAULT, 100, "getFirstGoods: ");
        MultiSearchResponse.Item[] responses = multiSearchResponse.getResponses();
        if (responses == null || responses.length == 0) {
            logger.error("the result of multiSearchResponse.getResponses()  is null or empty");
            return null;
        }
        Map<String, String> shopAndUrl = new HashMap<>();
        Arrays.stream(responses).forEach(response -> {
            SearchResponse internalResponse = response.getResponse();
            if (internalResponse == null) {
                logger.warn("internalResponse is null");
            } else {
                SearchHits hits = internalResponse.getHits();
                if (hits == null || hits.getTotalHits().value < 1) {
                    logger.warn("hits is null or empty");
                } else {
                    Map<String, Object> sourceAsMap = hits.getAt(0).getSourceAsMap();
                    Object shopId = sourceAsMap.get("shopId");
                    Object productImg = sourceAsMap.get("productImg");
                    if ((shopId instanceof String) && (productImg instanceof String)) {
                        shopAndUrl.put((String) shopId, (String) productImg);
                    }
                }
            }
        });

        if (shopAndUrl.isEmpty()) {
            logger.error("shopAndUrl is empty!!");
        }

        return shopAndUrl;
    }
}