如何在Python中对MongoDB数据按固定块大小计算平均值？

Hey there! Let's walk through exactly how to implement this chunked average calculation for your MongoDB records in Python. We'll start with the 2-record chunk example you mentioned, then scale it up to 10 records per chunk, and cover a couple of approaches depending on your needs.

Prerequisites #

First, make sure you have the pymongo library installed to interact with MongoDB:

pip install pymongo

Step 1: Connect to Your MongoDB Instance #

First, we'll set up the connection to your database and collection. Replace the placeholders with your actual database/collection names:

from pymongo import MongoClient
import math

# Connect to MongoDB (adjust the URI if you're using authentication or a remote server)
client = MongoClient('mongodb://localhost:27017/')
db = client['your_database']
source_collection = db['your_records']  # This is where your 10000 records live
result_collection = db['price_averages']  # We'll store averages here

Step 2: Basic Chunked Calculation (Using Skip & Limit) #

This approach works well if you want to know exactly how many chunks you'll have upfront. Let's start with your 2-record chunk example, then switch to 10.

Example: 2 Records Per Chunk #

chunk_size = 2  # Match your example first
total_records = source_collection.count_documents({})
total_chunks = math.ceil(total_records / chunk_size)

# Clear existing results (optional, depending on your needs)
result_collection.delete_many({})

for chunk_num in range(total_chunks):
    # Skip the records we've already processed, fetch the next chunk
    skip_count = chunk_num * chunk_size
    chunk_docs = list(source_collection.find({}, {'price': 1, '_id': 0}).skip(skip_count).limit(chunk_size))
    
    # Extract valid price values (skip docs without a price field)
    prices = [doc['price'] for doc in chunk_docs if 'price' in doc]
    
    if prices:  # Avoid division by zero if all docs in the chunk lack price
        avg_price = sum(prices) / len(prices)
        # Store the result with chunk metadata
        result_collection.insert_one({
            'chunk_number': chunk_num + 1,
            'average_price': round(avg_price, 2),  # Round for readability
            'records_processed': len(prices)
        })
        print(f"Chunk {chunk_num+1} done: Average price = {avg_price:.2f}")

Scale to 10 Records Per Chunk #

Simply change chunk_size = 10 and run the same code—everything else works the same way. For 10000 records, this will create 1000 chunks (10000 / 10 = 1000, no leftover records here).

Step 3: Streamlined Chunk Processing (Better for Large Datasets) #

If you're working with extremely large collections (bigger than 10k records), the skip() method can get slow as it has to traverse all previous records each time. Instead, we can process records as a stream, building chunks on the fly:

chunk_size = 10
current_chunk = []
chunk_number = 1

result_collection.delete_many({})

# Iterate through the cursor (MongoDB streams records efficiently)
for doc in source_collection.find({}, {'price': 1, '_id': 0}):
    if 'price' in doc:
        current_chunk.append(doc['price'])
    
    # When we hit the chunk size, calculate and store the average
    if len(current_chunk) == chunk_size:
        avg_price = sum(current_chunk) / chunk_size
        result_collection.insert_one({
            'chunk_number': chunk_number,
            'average_price': round(avg_price, 2),
            'records_processed': chunk_size
        })
        print(f"Chunk {chunk_number} done: Average price = {avg_price:.2f}")
        current_chunk = []
        chunk_number += 1

# Handle any remaining records that don't fill a full chunk
if current_chunk:
    avg_price = sum(current_chunk) / len(current_chunk)
    result_collection.insert_one({
        'chunk_number': chunk_number,
        'average_price': round(avg_price, 2),
        'records_processed': len(current_chunk)
    })
    print(f"Final chunk {chunk_number} done: Average price = {avg_price:.2f}")

This method is more efficient because it doesn't require counting records upfront or using skip(), which can be expensive for large datasets.

Notes & Edge Cases #

• Missing price fields: The code above skips documents without a price field. If you want to treat missing prices as 0 or another value, adjust the list comprehension to something like prices.append(doc.get('price', 0)).
• Data types: Ensure the price field is stored as a numeric type (int/float) in MongoDB. If it's a string, you'll need to convert it first (e.g., float(doc['price'])).
• Indexing: If you're fetching records in a specific order (e.g., sorted by _id), add an index on that field to speed up the cursor traversal.

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