Spark写入MongoDB速度慢（700万条数据耗时50分钟）求优化方案

• 调整Spark分区数
  分区数不合理会直接影响并行处理效率，建议让每个分区的记录数控制在1万-5万区间（700万数据对应140-700个分区，具体根据集群资源调整）。可以通过repartition调整：

  # 示例：设置200个分区
  dataframe = dataframe.repartition(200)
  dataframe.write.format("mongo").mode("append").options(**mongo_db_conn_options).save()
  

  提升分区并行度，能有效降低foreachPartition步骤的单分区处理压力。

• 优化MongoDB写入参数
  调整Connector的批量写入和一致性参数，减少写入等待时间：

  • 增大batchSize：设置批量写入的文档数（比如1000-5000，根据单文档大小调整，文档大则调小）
  • 降低写入一致性级别：如果业务允许，将writeConcern.w设为1（默认是majority，需要等待多数节点确认，耗时更长）
  • 关闭有序写入：设置ordered为false，允许并行写入分区，无需严格按顺序执行
    修改后的配置示例：

  mongo_db_conn_options = {
      "uri": self.MONGO_CONNECTION_URI.format(collection_name=collection_name),
      "database": self.MONGO_DB_CONFIG[MongoDBConstants.DB_NAME],
      "collection": collection_name,
      "batchSize": "5000",
      "writeConcern.w": "1",
      "ordered": "false"
  }
  

• 升级Spark MongoDB Connector版本
  新版本的Connector通常会优化批量写入逻辑，修复性能瓶颈，建议升级到最新稳定版，充分利用官方的性能优化。

• 调优Spark集群资源
  检查并调整Spark的executor资源配置：

  • 增加executor数量，提升整体并行能力
  • 合理分配executor的CPU和内存，比如设置--executor-cores 4 --executor-memory 8g --num-executors 10（根据集群实际资源调整），避免因资源不足导致的GC或处理缓慢。

• 匹配MongoDB分片键分区（分片集群场景）
  如果你的MongoDB是分片集群，将Spark DataFrame的分区键设置为MongoDB的分片键，这样每个Spark分区的数据会直接写入对应分片，避免跨分片的路由开销，大幅提升写入效率：

  # 假设分片键是user_id，按该字段哈希分区
  dataframe = dataframe.repartition(200, "user_id")
  

• 减少Schema转换开销
  提前对齐DataFrame和MongoDB集合的Schema，避免写入时的自动类型转换。如果DataFrame存在不必要的字段或类型不匹配，提前清理或转换，减少处理耗时。

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