在 MongoDB 中使用聚合管道存储前 N 个结果答案

【问题标题】：Bucketing top N results with an aggregation pipeline in MongoDB在 MongoDB 中使用聚合管道存储前 N 个结果
【发布时间】：2020-08-13 06:32:03
【问题描述】：

我正在使用 pymongo 对 MongoDB 进行一些分析。

在 MongoDB 中，有 480000 个 json 对象代表 2020 年 3 月至 2020 年 4 月之间关于 Covid-19 病毒的推文。

具体来说，这些对象包含两个字段：

1)"created_at"，表示推文的创建时间戳（例如 created_at : 2020-03-20T10:57:57.000+00:00 ），日期类型；

2)“retweet_count”，表示推文被转发的次数（例如“retweet_count:30”）；

我会创建一个聚合管道，每天接收前 5000 个 retweet_count 值最高的 json 对象。

问题是我不明白是否必须使用组子句、匹配子句或项目子句（我是新手）。

这里有一个我做过的尝试：

import pymongo
from datetime import datetime, tzinfo, timezone
from pymongo import MongoClient

client['Covid19']['tweets'].aggregate([
    {
        '$match' : {
            "created_at": { '$gte': datetime(2020, 3, 20), '$lt':  datetime(2020, 3, 21) } 
        }

    },
    {
        '$merge': {
            'into': 'tweets_filtered'
        }
    }
])

print(client['Covid19']['tweets_filtered'].count_documents({}))

这个管道提供了从 3 月 20 日到 3 月 21 日发布的推文，但我会概括这个过程，并每天获取前 5000 条 retweet_count 值最高的推文。

【问题讨论】：

标签： python mongodb pymongo pipeline

【解决方案1】：

您可以通过编程方式生成所需的边界并使用 $bucket。

$sort 输入以实现所需的排序（最先转发）。
$bucket 按天拆分集合。
$push 将每个文档移动到相应的日期。
$project 和 $slice 以获得前 X 个结果。

使用时间、计数和消息字段的 Ruby 示例：

require 'mongo'

Mongo::Logger.logger.level = Logger::WARN

client = Mongo::Client.new(['localhost:14420'])
c = client['foo']


c.delete_many

10.times do |i|
  day_time = Time.now - i*86400
  100.times do |j|
    time = day_time + j*100
    count = rand*1000
    message = "message #{count}"

    c.insert_one(time: time, count: count, message: message)
  end
end


days = (-1..10).map { |i| Time.now - i*86400 }.reverse
pp c.aggregate([
  {'$sort' => {count: -1}},
  {'$bucket' => {groupBy: '$time', boundaries: days,
    output: {messages: {'$push' => '$$ROOT'}},
  }},
  {'$project' => {top_messages: {'$slice' => ['$messages', 5]}}},
]).to_a

【讨论】：

【解决方案2】：

Pymongo 使用熊猫回答：

from pymongo import MongoClient
from datetime import datetime
import pandas as pd

TOP_N_PER_DAY = 5000

# Perform the find with a filter; strip out the _id
tweets = db.tweets.find({ 'created_at': {'$gte': datetime(2020, 3, 20), '$lt':  datetime(2020, 3, 22) }}, {'_id': 0})

# Create a dataframe from the find
df = pd.DataFrame(list(tweets))

# Convert the datetime to a date only timeseries
df['date'] = df['created_at'].dt.date

# Group by date and sort by retweet count
df = df.groupby('date').apply(lambda x: x.sort_values('retweet_count', ascending = False)).reset_index(drop=True)

# Take the top n per day
df = df.groupby('date').head(TOP_N_PER_DAY)

# Convert the pandas timeseries back to a datetime
df['date'] = pd.to_datetime(df['date'])

# Convert the dataframe into a list of dicts
records = df.to_dict('records')

# Insert the filtered tweets into a new collection
db.tweets_filtered.insert_many(records)

【讨论】：