springboot 整合 elasticsearch 三-----批量插入bulk和聚合统计

2023-10-27 08:41:58

基于第二章搭建的项目，下面我们实现es的批量插入

参考此文章，请先参考第二章进行搭建

springboot 整合 elasticsearch 二_q1035331653的博客-CSDN博客继续上一篇文章的内容项目整合elasticsearch如不清楚安装的，请看上一篇文章springboot 整合 elasticsearch 一安装elasticsearch_q1035331653的博客-CSDN博客环境说明:查询maven官网spring-boot 2.1.13对应的 SpringBootStarterDataElasticsearch»2.1.13.RELEASE对应的SpringDataElasticsearch»3.1.16.RELEASE对应的 elasticsear.https://blog.csdn.net/q1035331653/article/details/121907940实际项目开发中，用上es的场景，一般都有一下几个特征

1、海量数据的搜索（千万级别以上的数据）而且要求搜索响应要快

2、根据搜索条件返回的结果，有相等，类似，并且要求结果越接近的优先展示

3、在海量数据里的聚合统计（例如我开始的文章elasticsearch 聚合统计效率测试_q1035331653的博客-CSDN博客elasticsearch 的介绍就不啰嗦，大家可以上官网看。官网地址如下：What is Elasticsearch? | Elasticsearch Guide [7.16] | Elastic因为最近做的业务，数据结构比较复杂而且量大，因为查询聚合统计问题，最后考虑适用elasticsearch 提升业务功能查询统计效率。环境版本spring-boot-starter-parent 2.1.13.RELEASEspring-boot-starter-data-elahttps://blog.csdn.net/q1035331653/article/details/121848944）的场景，让我考虑使用es

牛逼吹多了！我们进入正文

同步数据到es 可以使用spring-boot-starter-data-elasticsearch 封装好的方法

saveAll(Iterable<S> entities)

看底层源码就知道也是调用了

elasticsearchOperations 的void bulkIndex(List<IndexQuery> queries);方法

当然灵活点可以自己写

这里我是自己写，每次插入的数量设置为15w

public class ElasticsearchUtil {

    /**
     * 批量插入的数据长度大小
     */
    private final int PAGE_SIZE = 150000;

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;


    /**
     * 批量插入数据
     * @return
     */
    public int bulkIndex(String indexName,String indexType,List<BaseEsEntity> dataList){
        int counter = 0;
        checkData(indexName,indexType,dataList);
        //判断索引是否存在
        if (!elasticsearchTemplate.indexExists(indexName)) {
            elasticsearchTemplate.createIndex(indexName);
        }
        List<IndexQuery> indexQueries = convertData(dataList,indexName,indexType);
        elasticsearchTemplate.bulkIndex(indexQueries);
        log.info("bulkIndex counter : " + indexQueries.size());
        counter = indexQueries.size();
        indexQueries.clear();
        dataList.clear();
        elasticsearchTemplate.refresh(indexName);

        return counter;
    }

    private void checkData(String indexName,String indexType,List<BaseEsEntity> dataList){
        if(StringUtils.isBlank(indexName) || StringUtils.isBlank(indexType)){
            throw new RuntimeException("indexName or indexType can not be null");
        }
        if(CollectionUtils.isEmpty(dataList)){
            throw new RuntimeException(String.format("保存的数据不能为空，data size 长度{%d}", dataList.size()));
        }
        if(dataList.size() > PAGE_SIZE){
            throw new RuntimeException(String.format("data size 必须小于{%d},当前长度{%d}", PAGE_SIZE, dataList.size()));
        }
    }

    private List<IndexQuery> convertData(List<BaseEsEntity> dataList,String indexName,String indexType){
        List<IndexQuery> queries = new ArrayList<>();
        for (BaseEsEntity esEntity : dataList) {
            IndexQuery indexQuery = new IndexQuery();
            indexQuery.setId(esEntity.getId());
            indexQuery.setSource(JSONObject.toJSONString(esEntity));
            indexQuery.setIndexName(indexName);
            indexQuery.setType(indexType);
            queries.add(indexQuery);
        }
        return queries;
    }

}

一下是聚合统计的代码展示

构建查询条件

BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
if (StringUtils.isNotBlank(request.getName())) {
    boolQueryBuilder.should(QueryBuilders.matchQuery(ElasticsearchDateKey.brandName, request.getName()));
    boolQueryBuilder.should(QueryBuilders.wildcardQuery(ElasticsearchDateKey.brandName+".keyword","*"+request.getName()+"*"));
    boolQueryBuilder.minimumShouldMatch("50%");
}
boolQueryBuilder.filter(QueryBuilders.termQuery(ElasticsearchDateKey.brandType,9));

NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
//查询条件
nativeSearchQueryBuilder.withQuery(boolQueryBuilder);
//分页
nativeSearchQueryBuilder.withPageable(pageable);
//2.2指定索引库和文档类型
nativeSearchQueryBuilder.withIndices(TRADEMARK_US_INDEX_NAME).withTypes(TRADEMARK_US_INDEX_TYPE);
int termsAggregationSize = request.getPageSize() * 10 > 10000 ? 10000 : request.getPageSize() * 10;
TermsAggregationBuilder field = AggregationBuilders.terms(ElasticsearchDateKey.typeCount).field(
        ElasticsearchDateKey.brandType).size(termsAggregationSize);
TermsAggregationBuilder applicationStatus = AggregationBuilders.terms(ElasticsearchDateKey.applicationStatus).field(
        ElasticsearchDateKey.applicationStatusKeyword).size(termsAggregationSize);
nativeSearchQueryBuilder.addAggregation(field);
nativeSearchQueryBuilder.addAggregation(applicationStatus);
//2.4构建查询对象
NativeSearchQuery nativeSearchQuery = nativeSearchQueryBuilder.build();
Page<TrademarkUsPagelistEO> byTitle = trademarkUs.search(nativeSearchQuery);

执行代码的debug结果

dsl查询结果

结果相符。

这里提一些个问题：代码通配符查询

QueryBuilders.wildcardQuery(ElasticsearchDateKey.brandName+".keyword","*"+request.getName()+"*")

这里增加了keyword 是不让 "*Ac*" 这样进行分词

问题二：分页查询出现-Result window is too large, from + size must be less than or equal to 的错误因为es默认的查询分页深度就是1w。正常情况下几乎没人从第一条数据查看到1w条，不正常的时候就是想看最后结果，杠精。遇到这样的产品提需求，请谁先k他一顿，并且问他是怎么做到产品的

解决办法：

PUT 索引库/_settings
{
"index":{
"max_result_window":1000000 #这个数值根据自身服务器资源或者需求的要求编写
}
}

码农公寓

相关文章