PySpark MongoDB连接器如何查询日期字段？返回空DataFrame怎么解决？

核心原因 #

你遇到的空DataFrame问题90%以上是因为数据类型不匹配：MongoDB中tweetInfo.created_at字段存储的是ISODate日期类型，你在PySpark查询中传入的是ISO格式字符串，Studio 3T会自动完成字符串到ISODate的隐式转换，但是PySpark的Mongo Spark Connector不会做该转换，导致条件匹配失败返回空结果。
其他可能的原因及对应解决方案如下： #

具体解决方案 #

1. 修正聚合查询的日期传参 #

不要将日期对象转为ISO字符串，直接传入带UTC时区的datetime对象即可：

import datetime

agg_query = [{
    '$match': {
        'tweetInfo.created_at': {
            '$gte': datetime.datetime(2021, 10, 10, 0, 0, 0, tzinfo=datetime.timezone.utc),
            '$lte': datetime.datetime.now(datetime.timezone.utc)
        }
    }
}]

注意必须添加UTC时区，否则本地时区的datetime对象和Mongo存储的UTC时区日期会出现时差，导致匹配结果不符合预期。

2. 手动指定读取Schema #

Spark默认采样部分数据推断Schema，若采样样本中created_at为空或格式异常，会导致字段类型推断错误。可以手动指定和Mongo结构匹配的Schema：

from pyspark.sql.types import StructType, StructField, StringType, TimestampType

custom_schema = StructType([
    StructField("_id", StringType(), nullable=True),
    StructField("tweetInfo", StructType([
        StructField("created_at", TimestampType(), nullable=True),
        # 补充其他需要查询的字段定义
    ]), nullable=True)
])

# 读取时指定schema
df = spark.read.format("mongo") \
    .schema(custom_schema) \
    .option("uri", "MongoDB连接URI") \
    .option("database", "数据库名") \
    .option("collection", "集合名") \
    .option("pipeline", agg_query) \
    .load()

3. 排查连接配置问题 #

先去掉聚合条件测试全量读取，确认连接配置正确：

test_df = spark.read.format("mongo") \
    .option("uri", "MongoDB连接URI") \
    .option("database", "数据库名") \
    .option("collection", "集合名") \
    .load()
# 输出总条数和前5条日期字段，确认数据正常读取
print(test_df.count())
test_df.select("tweetInfo.created_at").show(5, truncate=False)

如果全量读取也无数据，检查连接URI的权限、库名、集合名是否和Studio 3T中使用的一致。

4. 核对连接器版本兼容性 #

Mongo Spark Connector和MongoDB服务端存在版本匹配要求，比如MongoDB 5.0+需要使用10.x及以上版本的连接器，低版本连接器会出现查询异常。确认你的依赖版本符合官方兼容性要求即可。

旧版本连接器的聚合参数名为aggregation.pipeline，若使用3.x以下版本连接器，可以将.option("pipeline", agg_query)替换为.option("aggregation.pipeline", agg_query)测试。

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