当前位置：首页 > article >正文

MySQL5.7 百万数据迁移到 ElasticSearch7.x

article 2025/1/13 10:35:21

前言

在日常项目开发中，可能会遇到使用 ES 做关键词搜索的场景，但是一般来说业务数据是不会直接通过 CRUD 写进 ES 的。

因为这可能违背了 ES 是用来查询的初衷，数据持久化的事情可以交给数据库来做。那么，这里就有一个显而易见的问题：ES 里的数据从哪里来？

本文介绍的就是如何将 MySQL 的表数据迁移到 ES 的全过程。

一、一次性全量

该方案的思路很简单直接：将数据库中的表数据一次性查出，放入内存，在转换 DB 与 ES 的实体结构，遍历循环将 DB 的数据放入 ES 中。

但是对机器的性能考验非常大：本地 MySQL 10w 条数据，电脑内存16GB，仅30秒钟内存占用90%，CPU占用100%。太过于粗暴了，不推荐使用。

@Component05
@Slf4j
public class FullSyncArticleToES implements CommandLineRunner {

    @Resource
    private ArticleMapper articleMapper;

    @Resource
    private ArticleRepository articleRepository;

    /**
     * 执行一次即可全量迁移
     */
    //todo: 弊端太明显了，数据量一大的话，对内存和 cpu 都是考验，不推荐这么简单粗暴的方式
    public void fullSyncArticleToES() {
        LambdaQueryWrapper<Article> wrapper = new LambdaQueryWrapper<>();
        List<Article> articleList = articleMapper.selectList(wrapper);
        if (CollectionUtils.isNotEmpty(articleList)) {
            List<ESArticle> esArticleList = articleList.stream().map(ESArticle::dbToEs).collect(Collectors.toList());
            final int pageSize = 500;
            final int total = esArticleList.size();
            log.info("------------FullSyncArticleToES start!-----------, total {}", total);
            for (int i = 0; i < total; i += pageSize) {
                int end = Math.min(i + pageSize, total);
                log.info("------sync from {} to {}------", i, end);
                articleRepository.saveAll(esArticleList.subList(i, end));
            }
            log.info("------------FullSyncPostToEs end!------------, total {}", total);
        }
        else {
            log.info("------------DB no Data！------------");
        }
    }
    @Override
    public void run(String... args) {}
}

二、定时任务增量

这种方案的思想是按时间范围以增量的方式读取，比全量的一次性数据量要小很多。

也存在弊端：频繁的数据库连接 + 读写，对服务器资源消耗较大。且在极端短时间内大量数据写入的场景，可能会导致性能、数据不一致的问题（即来不及把所有数据都查到，同时还要写到 ES）。

但还是有一定的可操作性，毕竟可能没有那么极端的情况，高并发写入的场景不会时刻都有。

@Component
@Slf4j
public class IncSyncArticleToES {
    @Resource
    private ArticleMapper articleMapper;

    @Resource
    private ArticleRepository articleRepository;

    /**
     * 每分钟执行一次
     */
    @Scheduled(fixedRate = 60 * 1000)
    public void run() {
        // 查询近 5 分钟内的数据，有 id 重复的数据 ES 会自动覆盖
        Da

查看全文

http://www.kler.cn/a/229695.html