定期将MongoDB数据迁移至SQL Server的最佳方案及Apache Spark可行性咨询

最优方案选型 #

方案选择需优先满足不影响MongoDB线上高写入性能、数据一致性、同步效率、Schema兼容四个核心要求，分两种场景匹配最优解：

• 常规中小规模场景（单库数据量1TB以下，同步延迟要求15分钟以上），优先选择「MongoDB隐藏从节点读取 + Change Stream增量拉取 + 批量写入SQL Server + 数据校验」的轻量组合：
  • 读侧全部走MongoDB隐藏从节点，完全避免同步读压力挤占主节点的设备写入资源；增量数据通过Change Stream拉取上次同步完成后的变更记录，无需全表扫描；全量同步仅首次执行，后续按周/月按需补数即可
  • 转换层统一处理非结构化字段适配：MongoDB的嵌套文档、数组可以根据分析需求选择打平为SQL Server多表关联字段，或者直接存储为SQL Server原生JSON类型字段
  • 写入SQL Server优先用BULK INSERT接口，单批次提交量控制在1000-10000条（根据单条数据长度调整），避免单事务过大锁表
  • 调度层用常规作业调度工具触发即可，可根据需求配置同步周期，配套每次同步后的记录数、关键指标聚合结果校验逻辑，避免丢数
• 大数据量场景（单库1TB以上，有前置数据清洗、预处理需求），Apache Spark是非常成熟的选型，生产环境已有大量落地实践。

Apache Spark 场景落地说明 #

我们生产环境已经用Spark跑了2年完全一致的设备数据MongoDB迁SQL Server的作业，完全适配该场景，核心优势如下：

• 原生支持MongoDB、SQL Server官方连接器，无需开发底层读写逻辑，仅需配置连接参数即可
• 分布式计算能力强，TB级全量同步也能在小时级完成，不会出现单机同步的内存溢出问题
• 可直接在Spark任务中完成Schema转换、嵌套字段打平、去重、异常值清洗等逻辑，无需额外中转服务
• 增量同步支持两种实现：一是对接MongoDB Change Stream做准实时同步，二是基于Mongo集合的更新时间字段（需提前建索引）做过滤拉取，实现成本极低

核心代码示例（PySpark） #

需提前引入依赖：mongo-spark-connector、mssql-jdbc

from pyspark.sql import SparkSession

# 初始化Spark session
spark = SparkSession.builder \
    .appName("MongoToSQLServerSync") \
    .config("spark.mongodb.read.connection.uri", "mongodb://<Mongo隐藏从节点地址>/<库名>.<集合名>") \
    .getOrCreate()

# 增量读取：过滤大于上次同步时间的记录
df = spark.read.format("mongodb").load().filter("updateAt > '2024-05-01 00:00:00'")

# Schema转换：打平嵌套字段适配SQL Server表结构
df_transformed = df.select(
    df._id.alias("mongo_record_id"),
    df.deviceId.alias("device_id"),
    df.data.temperature.alias("device_temperature"),
    df.data.power.alias("device_power"),
    df.createAt.alias("create_time"),
    df.updateAt.alias("update_time")
)

# 批量写入SQL Server
df_transformed.write \
    .format("jdbc") \
    .option("url", "jdbc:sqlserver://<SQLServer地址>:1433;databaseName=<库名>") \
    .option("dbtable", "<目标表名>") \
    .option("user", "<用户名>") \
    .option("password", "<密码>") \
    .option("batchsize", 5000) \
    .mode("append") \
    .save()

生产踩坑提醒 #

• 绝对不要直接读MongoDB主节点做同步，避免读压力影响线上设备写入
• 增量同步用到的过滤字段（如updateAt）必须提前在MongoDB建索引，否则全表扫描会导致同步速度慢、从节点压力过高
• 写入SQL Server时不要用默认单分区，可根据设备ID、时间字段做数据分区，并行写入速度可提升数倍
• 大字段较多的场景需调优JDBC批量参数，避免Executor OOM

内容的提问来源于stack exchange，提问作者LearnItDom