Mongo Spark Connector v10.1对比v2.4性能骤降问题求助

问题背景 #

将Scala项目中的Mongo Spark Connector从v2.4升级到v10.1后，大数据量查询性能出现断崖式下滑：查询数亿条文档的MongoDB集合并转换为Dataset[T]，v2.4版本耗时约30分钟，v10.1版本耗时接近4小时（数据库未改动，其余条件一致）。

新旧版本核心代码对比 #

v2.4版本代码

def readAs[T <: Product : TypeTag](databaseName: String, collectionName: String,
                                     partitionerConfig: Option[MongoPartitionerConfig] = None): Dataset[T] = {
  import spark.implicits._
  val props = mongoProps(databaseName, collectionName, partitionerConfig)
  MongoSpark.load[T](spark, mongoReadConfig(props)).as[T]
}

v10.1版本适配代码

def readAs[T <: Product : TypeTag](databaseName: String, collectionName: String,
                                     partitionerConfig: Option[MongoPartitionerConfig] = None): Dataset[T] = {
  import spark.implicits._
  val props = mongoProps(databaseName, collectionName, partitionerConfig)
  val schema = ScalaReflection.schemaFor[T].dataType.asInstanceOf[StructType]

  spark.read.format("mongodb")
       .options(props)
       .schema(schema)
       .load()
       .as[T]
}

注：v10.1版本使用反射生成Schema是因为官方文档描述的自动推断功能失效，该操作解决了Schema匹配问题，但未解决性能问题；新旧版本均未使用自定义分区器。

可能的性能瓶颈点 #

1. 默认分区策略变更：v2.4与v10.1的默认分区器逻辑不同，v10.1可能默认使用低并行度的分区方式，导致Spark无法充分利用集群资源。
2. 谓词下推失效：新版连接器可能默认关闭了谓词下推优化，导致大量数据拉取到Spark端后再过滤，增加开销。
3. Schema处理开销：反射生成Schema或新版的数据序列化/反序列化逻辑存在额外性能损耗，大数据量下被放大。
4. 读取批次配置不合理：新版默认的MongoDB读取批次过小，导致多次请求数据库，累积延迟。

针对性解决建议 #

1. 强制配置高并行度分区策略 #

手动指定分区器并调整参数，确保Spark任务的并行度匹配集群资源：

// 在props中添加以下配置
props += ("spark.mongodb.read.partitioner" -> "com.mongodb.spark.sql.connector.read.partitioner.MongoPaginateBySizePartitioner")
props += ("spark.mongodb.read.partitioner.options.partitionSizeMB" -> "64") // 单分区数据大小，根据集群调整
props += ("spark.mongodb.read.partitioner.options.numberOfPartitions" -> "100") // 总分区数，建议为executor总core数的2-3倍

如果是MongoDB分片集群，可改用MongoShardedPartitioner进一步优化分区逻辑。

2. 启用谓词下推优化 #

确保查询条件下推至MongoDB端执行，减少拉取到Spark的数据量：

props += ("spark.mongodb.read.pushdown.filter" -> "true")
props += ("spark.mongodb.read.pipeline.includeFilters" -> "true")

3. 优化Schema处理 #

• 尝试开启新版的Schema自动推断功能（若可行）：

  props += ("spark.mongodb.read.schema.inference.enabled" -> "true")
  

• 若自动推断仍失效，手动定义StructType替代反射生成，减少运行时反射开销。

4. 调整读取批次大小 #

增大MongoDB读取批次，减少请求次数：

props += ("spark.mongodb.read.batchSize" -> "10000") // 根据MongoDB集群性能调整，建议5000-20000

5. 对比执行计划定位差异 #

对新旧版本的查询执行explain(true)，重点对比：

• 分区数量与分布逻辑
• 谓词是否下推至MongoDB端
• 数据扫描范围与过滤时机

6. 匹配Spark集群资源配置 #

确保Spark的executor-cores、num-executors与分区数匹配，避免资源闲置或过载：

• 总分区数建议为executor-cores * num-executors * 2~3

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