PySpark操作MongoDB聚合SQL的日期格式处理及报错问题

问题核心 #

你的问题集中在日期格式解析错误和聚合管道语法不合法两个点上，下面分别给出修正方案：

一、日期格式处理修正 #

第一种写法无效果是因为日期格式化字符串拼接错误，正确的格式应直接匹配+00:00时区部分，无需截取%f字段。

正确的日期解析方式 #

使用"%Y-%m-%dT%H:%M:%S.%f%z"作为格式化字符串，直接解析带UTC时区的日期：

from datetime import datetime

target_date_str = "2022-08-22T22:15:55.383+00:00"
parsed_date = datetime.strptime(target_date_str, "%Y-%m-%dT%H:%M:%S.%f%z")

如果需要匹配当天所有数据，可构造当天的UTC起始和结束时间：

# 当天0点UTC时间
start_date = datetime.strptime("2022-08-22T00:00:00.000+00:00", "%Y-%m-%dT%H:%M:%S.%f%z")
# 次日0点UTC时间（用于$lt条件）
end_date = datetime.strptime("2022-08-23T00:00:00.000+00:00", "%Y-%m-%dT%H:%M:%S.%f%z")

二、聚合管道语法修正 #

第二种写法报错是因为字典括号未闭合，且直接传递Python的datetime对象给Spark/MongoDB时，无法被正确序列化。

正确聚合管道写法（分场景） #

场景1：使用PyMongo直接操作MongoDB

from pymongo import MongoClient
from datetime import datetime

client = MongoClient("mongodb://localhost:27017")
db = client["entity"]
collection = db["status"]

target_date = datetime.strptime("2022-08-22T22:15:55.383+00:00", "%Y-%m-%dT%H:%M:%S.%f%z")

# 合法的聚合管道
pipeline = [
    {
        "$match": {
            "executionDate": {
                "$gte": target_date,
                "$lte": target_date
            }
        }
    }
    # 可添加其他聚合阶段（如$group、$project等）
]

result = list(collection.aggregate(pipeline))

场景2：使用PySpark连接MongoDB

需要用Spark的日期函数将字符串转换为Timestamp类型，避免Python对象序列化问题：

from pyspark.sql import SparkSession
from pyspark.sql.functions import to_timestamp

spark = SparkSession.builder \
    .appName("MongoDBAggregation") \
    .config("spark.mongodb.input.uri", "mongodb://localhost:27017/entity.status") \
    .getOrCreate()

df = spark.read.format("mongodb").load()

# 转换为Spark支持的Timestamp类型
target_ts = to_timestamp("2022-08-22T22:15:55.383+00:00", "yyyy-MM-dd'T'HH:mm:ss.SSSXXX")

# 执行过滤（后续可添加聚合操作）
filtered_df = df.filter(df.executionDate.between(target_ts, target_ts))
filtered_df.show()

替代方案：直接使用ISO日期字符串

如果MongoDB中executionDate字段存储为日期类型，直接传递ISO格式字符串也能被正确解析：

pipeline = [
    {
        "$match": {
            "executionDate": {
                "$gte": "2022-08-22T22:15:55.383+00:00",
                "$lte": "2022-08-22T22:15:55.383+00:00"
            }
        }
    }
]

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