Node.js项目中如何将Elasticsearch数据迁移至SQLite？

步骤1：批量导出 Elasticsearch 数据 #

使用 Elasticsearch 的 Scroll API 批量拉取数据（避免单次查询数据量过大导致超时），示例代码基于 @elastic/elasticsearch 客户端：

const { Client } = require('@elastic/elasticsearch');
const esClient = new Client({ node: 'http://localhost:9200' });

async function exportEsData(indexName) {
  let allData = [];
  // 初始化scroll查询
  const { body: scrollResp } = await esClient.search({
    index: indexName,
    scroll: '30s',
    size: 1000,
    query: { match_all: {} }
  });

  allData.push(...scrollResp.hits.hits);
  let scrollId = scrollResp._scroll_id;

  // 循环拉取剩余数据
  while (scrollResp.hits.hits.length > 0) {
    const { body: newScrollResp } = await esClient.scroll({
      scrollId,
      scroll: '30s'
    });

    if (newScrollResp.hits.hits.length === 0) break;
    allData.push(...newScrollResp.hits.hits);
    scrollId = newScrollResp._scroll_id;
  }

  // 清理scroll上下文，释放资源
  await esClient.clearScroll({ scrollId });
  // 返回原始业务数据（_source字段）
  return allData.map(item => item._source);
}

步骤2：数据格式映射转换 #

Elasticsearch 数据结构与 Sequelize 模型字段可能存在差异，需做字段名、数据类型的适配转换。示例：

// 假设Sequelize模型为User，需将ES数据映射为模型对应的字段结构
function transformEsData(esData) {
  return esData.map(item => ({
    id: item.es_unique_id, // 映射ES主键到SQLite主键
    username: item.user_account, // 字段名映射
    email: item.contact_email,
    createdAt: new Date(item.create_time), // ES日期字符串转JS Date
    updatedAt: new Date(item.update_time)
  }));
}

注意：若ES存在嵌套对象，可选择拆分为SQLite关联表，或直接用SQLite的JSON类型存储嵌套数据。

步骤3：批量插入 SQLite #

使用 Sequelize 的 bulkCreate 方法批量插入数据，分批次处理避免性能问题：

const { User } = require('./models'); // 导入你的Sequelize模型

async function importToSqlite(transformedData) {
  const batchSize = 1000; // 每批次插入行数，可根据性能调整
  for (let i = 0; i < transformedData.length; i += batchSize) {
    const batch = transformedData.slice(i, i + batchSize);
    await User.bulkCreate(batch, {
      ignoreDuplicates: true, // 主键重复时忽略（可选）
      updateOnDuplicate: ['username', 'email', 'updatedAt'] // 重复时更新指定字段（可选）
    });
    console.log(`已完成第 ${Math.floor(i/batchSize)+1} 批次，累计插入 ${i+batch.length} 条数据`);
  }
}

步骤4：整合执行迁移流程 #

将上述函数串联，完成全流程迁移：

async function runMigration() {
  try {
    const esData = await exportEsData('your_es_index_name'); // 替换为你的ES索引名
    console.log(`成功导出 ${esData.length} 条ES数据`);
    
    const transformedData = transformEsData(esData);
    console.log('数据格式转换完成');
    
    await importToSqlite(transformedData);
    console.log('全量数据迁移完成');
  } catch (error) {
    console.error('迁移失败:', error.message);
  }
}

runMigration();

关键注意事项 #

• 大数量迁移时，建议增加断点续传机制（比如记录已处理的最后一条数据ID），避免中途失败需重新全量执行。
• 迁移前先做小批量测试，验证数据转换逻辑和插入结果的正确性。
• 若迁移期间ES仍有写入，需在全量迁移后做增量同步（比如基于时间戳拉取新增/更新数据）。
• 注意SQLite的存储限制，单表数据量过大时需考虑分表优化。

内容的提问来源于stack exchange，提问作者Sameh.K