如何在PySpark中对MongoDB执行find查询而非全量加载集合？

当然可以实现，核心是利用MongoDB Spark Connector的查询下推能力，让过滤逻辑直接在MongoDB端执行，不用全量加载集合后再在PySpark侧处理。以下是两种实用的实现方法：

方法一：通过PySpark过滤语法自动转换为MongoDB查询 #

MongoDB Spark Connector会自动识别PySpark的过滤条件，将其转换为MongoDB原生的find查询，直接在数据库端完成数据筛选。

步骤示例： #

1. 初始化配置好MongoDB连接的SparkSession：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("DirectMongoFind") \
    .config("spark.mongodb.input.uri", "mongodb://<host>:<port>/<database>.<collection>") \
    .config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.12:10.2.0")  # 版本需匹配你的Spark和MongoDB
    .getOrCreate()

2. 读取数据并添加过滤条件：

# 此时不会全量加载集合数据
df = spark.read.format("mongodb").load()

# 编写过滤条件，连接器会自动转为MongoDB的find查询在端执行
filtered_df = df.filter("age > 30 AND city = 'Shanghai'")

# 验证查询是否下推：查看执行计划，若包含`PushDownFilter`则说明条件在MongoDB端处理
filtered_df.explain()

方法二：直接传递MongoDB原生查询文档 #

如果需要复杂的原生查询逻辑（比如嵌套条件、数组匹配等），可以通过pipeline参数直接传递MongoDB的查询语句，完全复用原生find的语法。

步骤示例： #

1. 导入模块并初始化SparkSession（同方法一）：

import json
from pyspark.sql import SparkSession

# 初始化SparkSession代码同方法一

2. 定义原生查询并传递给连接器：

# 对应MongoDB原生查询：db.collection.find({"age": {"$gt": 30}, "tags": {"$in": ["tech", "finance"]}})
mongo_find_query = {
    "$match": {
        "age": {"$gt": 30},
        "tags": {"$in": ["tech", "finance"]}
    }
}

# 将查询转为JSON字符串，通过pipeline参数传递
df = spark.read.format("mongodb") \
    .option("pipeline", json.dumps([mongo_find_query])) \
    .load()

关键注意事项： #

• 版本兼容：确保MongoDB Spark Connector版本与你的Spark（2.x/3.x）、MongoDB服务器版本匹配，避免兼容性问题。
• 避开无法下推的操作：如果使用PySpark自定义UDF或连接器不支持的函数，会触发全量加载后过滤，尽量用内置的可下推函数。
• 执行计划验证：始终用df.explain()确认查询是否被下推，防止意外的全量加载。

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