Spark与MongoDB连接器聚合查询参数化配置问题咨询

该报错是由于参数化拼接后的pipeline格式不符合Mongo Spark连接器的要求，常见诱因包括：引号不匹配、日期格式非法、聚合阶段没有用数组包裹、特殊字符未正确转义。

推荐优先使用json序列化方式生成pipeline，可完全避免手动拼接的格式错误，也可根据需求选择f-string拼接方案，完整可运行代码如下：

from pyspark.sql.functions import *
from datetime import datetime, timedelta
import json

# 计算90天前的日期，转换为MongoDB ISODate兼容的ISO 8601格式字符串
cutoff_date = (datetime.now() - timedelta(days=90)).strftime("%Y-%m-%dT%H:%M:%SZ")

# 方案1：json序列化生成合法pipeline（推荐，无需手动处理转义）
pipeline = json.dumps([
    {
        "$match": {
            "timestamp": {
                "$gte": {"$date": cutoff_date}
            }
        }
    }
])

# 方案2：f-string手动拼接方案，注意大括号用双写转义、引号匹配
# pipeline = f"[{{'$match': {{'timestamp': {{$gte: ISODate('{cutoff_date}')}}}}}}]"

df = (spark.read
           .format("com.mongodb.spark.sql.DefaultSource") 
           .option("uri", connectionString)
           .option("database", 'my database')
           .option("collection", 'my collection')
           .option("pipeline", pipeline)
           .load())

• 所有版本的Mongo Spark连接器都要求pipeline参数为JSON数组格式，哪怕只有一个聚合阶段，也需要用[]包裹，否则会直接抛出无效pipeline的错误
• 传入ISODate的日期必须符合ISO 8601标准格式，示例中%Y-%m-%dT%H:%M:%SZ的输出格式完全匹配MongoDB的解析要求
• 使用json.dumps生成pipeline可以避免手动拼接时的单双引号冲突、大括号转义错误问题，更稳定易维护

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