Spark DataFrame过滤MongoDB数据结果不一致问题求助

你遇到的这个问题挺典型的——处理较大规模MongoDB集合（26000条）时，Spark DataFrame过滤后的计数结果不稳定且低于实际值，但小数据量（5000条）正常，仅缓存操作能解决问题。结合你的环境（Spark 2.2.0、MongoDB 3.4、本地模式），我来拆解下可能的原因和对应的解决方案：

可能的原因 #

1. Mongo Spark连接器的分区配置错误（核心原因） #

你尝试修改分区策略但未生效，是因为配置参数用错了：spark.mongodb.input.partitioner才是指定分区器的参数，而你把分区器名称写到了partitionerOptions里，导致分区策略根本没生效。

Spark 2.2.0搭配的旧版Mongo连接器，默认分区策略在处理大集合时，容易出现分区数据边界遗漏或部分分区读取不完整的情况。虽然第一次count()能拿到正确结果（全量扫描不依赖分区过滤），但过滤操作触发分区级扫描时，异常的分区划分会导致部分数据被遗漏，结果自然不稳定。

2. 旧版本连接器的过滤下推Bug #

Spark 2.2.0对应的Mongo Spark连接器版本（如mongo-spark-connector_2.11:2.2.0）比较老旧，存在全匹配过滤条件下推异常的Bug：当过滤条件是全集合都满足的字段时，连接器的下推逻辑会出现错误，导致MongoDB返回的结果集不完整。而缓存操作会把全量数据加载到内存，过滤时直接在内存执行，绕开了下推的问题，所以结果正确。

3. 本地模式的资源竞争 #

你用的是本地8核16G配置，Spark Executor和MongoDB客户端可能存在IO资源竞争：处理大集合时，部分分区的读取任务会因为IO阻塞导致数据读取不完整，而Spark的容错机制没有正确重试这些任务，最终导致计数结果偏小且波动。

解决方案 #

1. 修正分区器配置 #

正确配置MongoPaginateBySizePartitioner，让分区划分更均匀，避免数据遗漏：

val readConfig = ReadConfig(Map(
  "collection" -> collectionName,
  "spark.mongodb.input.readPreference.name" -> "primaryPreferred",
  "spark.mongodb.input.database" -> dataBaseName,
  "spark.mongodb.input.uri" -> hostName,
  // 指定分区器类路径（旧版连接器用com.mongodb.spark.sql.partitioner.MongoPaginateBySizePartitioner）
  "spark.mongodb.input.partitioner" -> "com.mongodb.spark.sql.connector.read.partitioner.MongoPaginateBySizePartitioner",
  "spark.mongodb.input.partitionerOptions.partitionSizeMB" -> "64" // 可根据数据大小调整，默认64MB
))
val dataFrame = MongoSpark.load(spark, readConfig)
dataFrame.filter(dataFrame.col("column_1") === "column_1_value").count()

2. 临时禁用过滤下推 #

如果分区配置调整后仍有问题，可以关闭过滤下推，让逻辑在Spark端执行，绕开连接器的Bug：

// 单DataFrame禁用
val dataFrame = MongoSpark.load(spark, readConfig)
  .hint("DISABLE_PUSH_DOWN_FILTERS")
dataFrame.filter(dataFrame.col("column_1") === "column_1_value").count()

// 全局禁用（在SparkSession初始化时配置）
val spark = SparkSession.builder()
  .appName("MongoSparkExample")
  .config("spark.mongodb.input.pushdown.filter", "false")
  .getOrCreate()

3. 升级Spark与连接器版本（长期解决方案） #

Spark 2.2.0是2017年的老旧版本，配套的Mongo连接器存在大量已知Bug。建议升级到兼容的新版本：

• Spark升级到3.1.x或3.2.x系列
• 对应Mongo Spark连接器升级到mongo-spark-connector_2.12:10.0.x（适配MongoDB 3.4+和Spark 3.x）
  新版本在分区管理、过滤下推逻辑上有大幅优化，能从根本上解决这类稳定性问题。

4. 调整本地模式资源配置 #

限制Spark核心数，避免和MongoDB客户端抢IO资源：

val spark = SparkSession.builder()
  .appName("MongoSparkExample")
  .master("local[4]") // 用4核而非8核，留出资源给MongoDB
  .config("spark.driver.memory", "8g")
  .config("spark.executor.memory", "8g")
  .getOrCreate()

验证建议 #

优先尝试修正分区器配置，这是最直接的解决办法；如果无效再禁用过滤下推；长期来看，升级版本是彻底解决问题的最优方案。

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