Elasticsearch迁移数据方案（不停服重建索引）

背景

• Elasticsearch是一个高扩展的开源全文搜索和分析引擎，它允许存储、搜索和分析大量的数据。
• ES 的索引创建之后的 mapping 结构是不能够修改的，如果系统中的 ES 索引没有使用别名指向真正的索引，那么要达到增删mapping字段的需求，就需要对索引进行重建。

方案

流程图

image

打开临时索引

• 在系统中定义开关，读取外部配置文件，当开关置为 true 时，所有数据将交给临时索引保存，也就是index_temp 进行处理。
• 在index_temp接收数据的过程中，index_old依旧可以提供查询服务，但是之前程序如果使用的是index_old查询的话，会造成新添加进 index_temp 的数据不可查询

拷贝数据

• 将index_old的数据拷贝到index_new中

 ReindexAction.INSTANCE
 .newRequestBuilder(elasticsearchTemplate.getClient())
                    .source("index_old")
                    .destination("index_new")
                    .get();

删除老索引&建立别名

• 目前index_old中的数据已经全部拷贝到index_new中了，现在就要把index_old删掉，建立index_old的别名指向index_new，也就是说将 index_old 删掉，再建立名叫index_old的别名指向 index_new, 此时index_old 和 index_new 都能提供服务

  if (elasticsearchTemplate.indexExists("index_old")) {
            elasticsearchTemplate.deleteIndex("index_old");
        }

        AliasQuery aliasQuery = new AliasQuery();
        aliasQuery.setIndexName("index_new");
        aliasQuery.setAliasName("index_old");
        elasticsearchTemplate.addAlias(aliasQuery);

关闭临时索引

• 将开关置为 false，此时所有数据将保存到index_new， 就算程序中使用的名字是 index_old，数据也会保存到 index_new, 因为设置了别名指向 index_new

迁移临时索引数据到新索引

int curPage = 0;
    while (true) {
        SearchQuery query = new NativeSearchQuery(QueryBuilders.matchAllQuery()).setPageable(PageRequest.of(curPage, 1000));
        Page<IndexTemp> indexTemp = elasticsearchTemplate.queryForPage(query, IndexTemp.class);
            if (!indexTemp.hasContent()) {
                logger.info("index_temp 没有跟多数据");
                logger.info("index_temp 迁移完成");
                 return;
              }
             List<IndexNew> infoList = new ArrayList<>();
            for (IndexTemp temp : indexTemp) {
                   IndexNew indexNew = new IndexNew();
                BeanUtils.copyProperties(temp, indexNew);
                   infoList.add(indexNew);
             }
             List<IndexQuery> indexQueries = infoList.stream().map(e -> {
                  IndexQuery indexQuery = new IndexQuery();
                 indexQuery.setObject(e);
                 return indexQuery;
              }).collect(Collectors.toList());

            elasticsearchTemplate.bulkIndex(indexQueries);
              logger.info("迁移一页, page:{}, infoListSize:{}", curPage, infoList.size());
             curPage++;
         }

删除临时索引

if (elasticsearchTemplate.indexExists("index_temp")) {
            elasticsearchTemplate.deleteIndex("index_temp");
            logger.info("index_temp 删除完成");
        }

总结

最后还是建议大家使用别名来指向真正的索引，使用别名对于索引结构的变更非常方便，只要将更改别名指向就 OK 了，简单快捷