如何在PySpark中通过单连接访问MongoDB多数据库与集合

需求说明 #

已建立MongoDB连接，该连接下包含多个数据库与集合，希望在PySpark中通过单连接复用的方式查询不同集合，避免每次查询新建连接导致性能损耗。现有代码每次查询都重复配置完整的读取参数，需要实现可灵活切换数据库、集合及查询语句的复用方式。

优化方案 #

通过封装复用基础的DataFrameReader配置，仅在每次查询时修改差异化参数（数据库、集合、查询管道）即可。同时MongoDB Spark Connector本身会维护连接池，确保连接复用，无需手动管理连接生命周期。

代码实现

# 预先配置公共连接参数，复用基础读取器
base_reader = spark.read.format("com.mongodb.spark.sql.DefaultSource") \
    .option("uri", "string_connection")

# 封装查询函数，灵活传入差异化参数
def read_mongo(database, collection, pipeline):
    return base_reader \
        .option("database", database) \
        .option("collection", collection) \
        .option("pipeline", pipeline) \
        .load()

# 调用示例
database_1 = "data_1" 
database_2 = "data_2"
collection_1 = "client_1"
collection_2 = "client_2"
myquery_1 = [query_1]
myquery_2 = [query_2]

dataframe_1 = read_mongo(database_1, collection_1, myquery_1)
dataframe_2 = read_mongo(database_2, collection_2, myquery_2)

关键说明 #

• base_reader预先固定了连接URI，后续查询只需覆盖database、collection和pipeline参数，无需重复初始化完整读取器，减少冗余代码。
• MongoDB Spark Connector内部自动维护连接池，所有基于同一URI的查询都会复用池内连接，从根源避免重复建连接的性能损耗。
• 封装函数后可快速扩展查询场景，新增查询只需传入对应参数即可，代码更简洁易维护。

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