spring-data-es 基本操作使用

Posted 2022-05-27 wujiumengxing

索引创建过程,以及前置的Dao等等实现和配置略过….

背景:

本地创建了一个名为User的索引 结构如下

/**
 * 单条插入数据
 */
@Test
public void save() 
    User user = new User();
    user.setId(1L);
    user.setImage("wwwwwwww.111111");
    user.setName("zhuxueyu");
    user.setPrice(999.99);
    user.setTitle("none");
    user.setBirthday(toESTime(LocalDateTime.now()));
    userDao.save(user);


/**
 * 批量数据插入
 */
@Test
public void saveAll() 
    List<User> list = new ArrayList<>();
    User user = new User();
    user.setId(2L);
    user.setImage("wwwwwwww.2222222");
    user.setName("张杰");
    user.setPrice(1999.99);
    user.setTitle("杰哥");
    user.setBirthday(toESTime(LocalDateTime.now()));
    User user2 = new User();
    user2.setId(3L);
    user2.setImage("wwwwwwww.333333");
    user2.setName("lixiaopeng");
    user2.setPrice(2999.99);
    user2.setTitle("鹏哥");
    user2.setBirthday(toESTime(LocalDateTime.now()));
    list.add(user);
    list.add(user2);

    userDao.saveAll(list);


/**
 * 更新数据
更新数据(其实和插入是一个方法只是他会数据覆盖 必须要传入id和其他数据值 如果只传入你想改的数据值那么 最终结果只会留下id和你传入的值) 
 *
 * @param
 * @return
 */
@Test
public void update() 
    User user = new User();
    user.setId(5555L);
    user.setImage("wwwwwwww.111111");
    user.setName("zhuxueyu");
    user.setPrice(9999.99);
    user.setTitle("朱");
    userDao.save(user);


/**
 * 批量更新
 */
@Test
public void updateAll() 
    List<User> list = new ArrayList<>();
    User user = new User();
    user.setId(2L);
    user.setImage("wwwwwwww.2222222");
    user.setName("张杰");
    user.setPrice(5999.99);
    user.setTitle("杰哥");
    user.setBirthday(toESTime(LocalDateTime.now()));
    User user2 = new User();
    user2.setId(3L);
    user2.setImage("wwwwwwww.333333");
    user2.setName("张杰");
    user2.setPrice(6999.99);
    user2.setTitle("鹏哥");
    user2.setBirthday(toESTime(LocalDateTime.now()));
    list.add(user);
    list.add(user2);
    userDao.saveAll(list);


/**
 * 部分数据更新

部分udpdate似乎不ElasticsearchRepository支持。所以我用ElasticsearchTemplate

需要集成一下ElasticsearchTemplate

 */
@Test
public void updateSome() 
    UpdateRequest updateRequest = new UpdateRequest();
    try 
        updateRequest.doc(jsonBuilder().startObject()
                .field("price", "5999.99")
                .endObject());
        UpdateQuery build2 = new UpdateQueryBuilder().withId("3").withClass(User.class).withUpdateRequest(updateRequest).build();
        elasticsearchRestTemplate.update(build2);
     catch (IOException e) 
        e.printStackTrace();
    


/**
 * 通过脚本进行数据查询并更新

updateByQuery  是根据你的搜索条件检索出来的结果进行更新操作处理,同时他也支持版本控制操作-> updateByQueryRequest.setRefresh(true); 这个东西进行限制的;不设置的话不会进行版本控制.

支持script的语法为painless 具体链接地址:

Painless API Reference | Elasticsearch Reference [6.2] | Elastic

 */
@Test
public void updateByQueryTest() 
    // 构造方法指定index_name
    UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest("user");
    updateByQueryRequest.setRefresh(true);
    // 指定修改数据的搜索条件
    BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.termQuery("id", 2L));
    updateByQueryRequest.setQuery(queryBuilder);
    //设置版本冲突时继续
    updateByQueryRequest.setConflicts("proceed");
    StringBuilder sb = new StringBuilder();
    Map<String, Object> map = new HashMap<>();
    map.put("name", "张杰啊啊啊啊啊啊啊");
    map.keySet().stream().forEach(p -> 
        sb.append("ctx._source.").append(p).append("=params.").append(p).append(";");
    );
    Script script = new Script(ScriptType.INLINE, Script.DEFAULT_SCRIPT_LANG, sb.toString(), map);
    updateByQueryRequest.setScript(script);
    try 
         elasticsearchRestTemplate.getClient().updateByQuery(updateByQueryRequest, RequestOptions.DEFAULT);
     catch (IOException e) 
        e.printStackTrace();
    


@Test
public void deleteById() 
    userDao.deleteById(5555L);


/**
 * Run async.
 *
 * @param runnable the runnable
 */
public static void runAsync(Runnable runnable) 
    CompletableFuture.runAsync(runnable)
            .exceptionally(ex -> 
                log.error("", ex);
                return null;
            );


/**
 * 分页查询
 *
 * @param
 * @return
 */
@Test
public void getPageList() 
    NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
    Pageable page = PageRequest.of(0, 10);
    nativeSearchQueryBuilder.withPageable(page);
    BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
    // boolQueryBuilder.must(QueryBuilders.matchQuery("name","张杰"));
    //  nativeSearchQueryBuilder.withQuery(boolQueryBuilder);
    //nativeSearchQueryBuilder.withSort(SortBuilders.fieldSort("id").order(SortOrder.DESC));
    Iterable<User> search = userDao.search(nativeSearchQueryBuilder.build());
    Iterator<User> iterator = search.iterator();
    while (iterator.hasNext()) 
        User next = iterator.next();
        System.out.println(next);
    


/**
 * 滚动查询(普通查询实时性比较高的情况且数据量不大,因为数据量大了他的查询最后一页数据时是把之前所有页的数据全部都加载到内存 然后最后处理一下然后把10条返还给我们.滚动查询适合大量数据搜索,一批一批的查询,保存一个当时内存快照,且有一个时间过期范围,这样可以有效提升性能)
 */
@Test
public void scroll() 
    List<User> list = new ArrayList<>();
    SearchQuery searchQuery = new NativeSearchQueryBuilder().build();
    ScrolledPage<User> scroll = elasticsearchRestTemplate.startScroll(10000, searchQuery, User.class);
    String scrollId = null;
    while (scroll.hasContent()) 
        list.addAll(scroll.getContent());
        scrollId = scroll.getScrollId();
        scroll = elasticsearchRestTemplate.continueScroll(scrollId, 10000, User.class);
    
    // 进行数据清楚 如果这不清除就会到你设置的过期时间后才会自动清除
    elasticsearchRestTemplate.clearScroll(scrollId);
    System.out.println(list.toString());


/**.查询聚合操作之term 这个功能类似mysql的group by
 * 分组查询
 */
@Test
public void termQuery() 
    NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder()
            .addAggregation(AggregationBuilders.terms("aname")
                    .field("name")
                    .size(100)
            );
    AggregatedPage<User> aggPage
            = (AggregatedPage<User>) userDao.search(queryBuilder.build());
    Aggregation aname = aggPage.getAggregation("aname");
    ParsedStringTerms aa = (ParsedStringTerms) aname;
    List<? extends Terms.Bucket> buckets = aa.getBuckets();
    for (Terms.Bucket bucket : buckets) 
        System.out.println(bucket.getKey());
    


/**
 * 查询最大值
 */
@Test
public void maxQuery() 
    MaxAggregationBuilder field = AggregationBuilders.max("maxPrice").field("price");
    NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
    nativeSearchQueryBuilder.addAggregation(field);
    Aggregations query = elasticsearchRestTemplate.query(nativeSearchQueryBuilder.build(), SearchResponse::getAggregations);
    Aggregation maxPrice = query.get("maxPrice");
    System.out.println(((ParsedMax) maxPrice).getValue());


/**
 * 查询最小值
 */
@Test
public void minQuery() 
    MinAggregationBuilder field = AggregationBuilders.min("minPrice").field("price");
    NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
    nativeSearchQueryBuilder.addAggregation(field);
    Aggregations query = elasticsearchRestTemplate.query(nativeSearchQueryBuilder.build(), SearchResponse::getAggregations);
    Aggregation maxPrice = query.get("minPrice");
    System.out.println(((ParsedMin) maxPrice).getValue());


/**
 * 求条数
 */
@Test
public void getCount() 
    ValueCountAggregationBuilder field = AggregationBuilders.count("peopleCount").field("id");
    NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
    nativeSearchQueryBuilder.addAggregation(field);
    Aggregations query = elasticsearchRestTemplate.query(nativeSearchQueryBuilder.build(), SearchResponse::getAggregations);
    ValueCount peopleCount = query.get("peopleCount");
    System.out.println(peopleCount.getValue());



public static String toESTime(LocalDateTime localDateTime) 
    if (null == localDateTime) 
        return null;
    
    return localDateTime.atZone(ZoneId.systemDefault()).format(ES_DATE_TIME);


private static final DateTimeFormatter ES_DATE_TIME = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZZ");