PySpark读取MongoDB大集合过滤查询耗时过长的配置问题

你的判断完全正确，默认配置下Mongo Spark Connector不会将Spark侧的过滤、聚合算子下推到MongoDB执行，会全量拉取整个集合的数据到Spark集群后再做计算，大集合场景下会出现Driver端分片采样耗时长、任务长时间不展示、整体执行极慢的问题，小集合因为数据量小全量拉取开销低所以表现正常。

需要补充以下核心配置，让Connector利用MongoDB已有的索引执行过滤，避免全表扫描：

• 必须开启算子下推配置，这是解决性能问题的核心

  # 开启聚合下推，支持count、group by等聚合操作直接在Mongo侧执行
  .set("spark.mongodb.read.pushDownAggregation", "true")
  # 开启过滤条件下推，自动将where中的等值、范围匹配转换为MongoDB查询语句
  .set("spark.mongodb.read.pushDownFilters", "true")
  # 开启limit下推，SQL中的LIMIT语法会直接传递给MongoDB，不会拉取全量数据后截断
  .set("spark.mongodb.read.pushDownLimit", "true")
  

• 优化分区策略，避免大集合下默认采样切分耗时过长

  # 选择有索引的高频过滤字段作为分区键，你的查询均带TimeStamp过滤且该字段有索引，直接指定为TimeStamp
  .set("spark.mongodb.read.partitionKey", "TimeStamp")
  # 根据集群核数设置合理分区大小，你当前总计算核数为16，设置64MB单分区大小可保证并行度充足
  .set("spark.mongodb.read.partitionSizeMB", "64")
  

1. 配置调整后可以通过df.filter((df["TimeStamp"]>1650050989) & (df["TimeStamp"]<1655555989)).explain(True)查看物理计划，若PushedFilters列表中包含你写的GreaterThan、LessThan、EqualTo等过滤条件，说明下推已生效，MongoDB会直接走对应索引返回匹配数据。
2. 若使用2.x版本的Mongo Spark Connector，上述配置的前缀需要从spark.mongodb.read.替换为spark.mongodb.input.。
3. 确保查询时字段类型匹配：例如Mongo中TimeStamp存储为整型时，Spark侧传参也要用整型，类型不匹配会导致下推失效，退回全表扫描。
4. 额外核对你的库表配置：当前配置中spark.mongodb.read.database设置为collection_1，请确认该值为实际的MongoDB库名，ADPacket为库下的集合名，避免库表配置错误导致的异常。

配置生效后，你执行的两类查询都会直接命中MongoDB索引：

• 带AssetConnectDeviceKey等值+TimeStamp范围的查询会直接命中对应字段的单字段/复合索引，返回匹配的前500条数据，不会扫描全表
• 范围count操作会直接在MongoDB侧通过索引统计符合条件的文档数，不需要拉取全量数据到Spark计算，耗时会从小时级降到秒级

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