PySpark MongoDB查詢日期

Question

我在MongoDB中使用PySpark並希望使用帶日期過濾器的管道查詢我的數據庫。 在蒙戈我的查詢看起來像這樣：

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讀者期待值，但發現'日期時間'。

（我想要做的一切的管道，而不是在一個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()