python - PySpark MongoDB 查询日期

Question

我将 PySpark 与 MongoDB 结合使用，并希望使用带有日期过滤器的管道查询我的数据库。在 Mongo 中，我的查询看起来像这样:

db.collection.aggregate([{$match:{"creation":{$lte:new Date("Jan 1, 2016")}}},{$sort:{"creation":1}}])

但我不知道如何在 Python 中做同样的事情。例如我试过:

pipeline = [{'$match': {'creation': {'$lte': datetime.datetime(2016, 1, 1, 0, 0)}}}, {'$sort': {'creation': 1}}]
df = context.read.format("com.mongodb.spark.sql").options(pipeline=pipeline).load()

我收到一个错误:org.bson.json.JsonParseException:JSON 阅读器需要一个值但找到了“datetime”。

(我想在管道中而不是在 SQL 查询中执行所有操作)

Answer 1

您可以利用 MongoDB extended JSON指定日期。例如:

pipeline = [{'$match':{'creation':{'$lte': {'$date': "2016-01-01T00:00:00Z" }}}}]
df_pipeline = sqlContext.read.format("com.mongodb.spark.sql.DefaultSource")
                             .option("pipeline", pipeline).load()
df_pipeline.first()