Hướng dẫn Aggregation Pipeline của MongoDB bằng Python với PyMongo

Làm việc với các bộ dữ liệu lớn thường đặt ra thách thức là trích xuất các mẫu và insight có ý nghĩa trong khi vẫn duy trì hiệu năng. Khi ứng dụng của bạn lưu trữ dữ liệu trong MongoDB, chạy các truy vấn và phép biến đổi phức tạp trực tiếp trong cơ sở dữ liệu có thể nhanh hơn nhiều so với việc chuyển dữ liệu sang các công cụ phân tích bên ngoài. Aggregation pipeline của MongoDB mang đến giải pháp bằng cách cho phép bạn xử lý, biến đổi và phân tích dữ liệu ngay tại nơi nó cư trú.

Bạn có thể tạo các quy trình xử lý dữ liệu tùy chỉnh bằng cách nối các thao tác đơn giản theo trình tự. Mỗi stage trong pipeline biến đổi document và truyền kết quả sang stage tiếp theo. Chẳng hạn, bạn có thể cần lọc bản ghi theo khoảng thời gian, nhóm theo danh mục, tính các thước đo thống kê và định dạng đầu ra — tất cả được thực hiện qua một thao tác cơ sở dữ liệu duy nhất xử lý dữ liệu sát nguồn.

Trong bài viết này, bạn sẽ học cách xây dựng aggregation pipeline để giải quyết các thách thức dữ liệu phổ biến, với tất cả ví dụ minh họa bằng PyMongo (client Python chính thức của MongoDB). 

Mặc dù bài viết tập trung vào aggregation pipeline, nếu bạn mới dùng MongoDB với Python, khóa học Introduction to MongoDB in Python là điểm khởi đầu toàn diện. Từ bài này, bạn sẽ có đủ nền tảng để tự chuyển các khái niệm aggregation sang ngôn ngữ truy vấn của MongoDB hoặc với sự hỗ trợ của mô hình ngôn ngữ.

Aggregation Pipeline trong MongoDB là gì?

Hãy hình dung bạn cần phân tích đánh giá của khách hàng trên nhiều sản phẩm để hiểu xu hướng hài lòng. Các truy vấn truyền thống có thể lấy dữ liệu, nhưng chúng không giúp kết hợp, phân tích và biến đổi thông tin đó thành các tóm tắt hữu ích. 

Aggregation pipeline của MongoDB giải quyết điều này bằng cách cung cấp một cách có cấu trúc để xử lý dữ liệu qua một chuỗi thao tác xây dựng trên nhau.

Kiến trúc pipeline và luồng document

Hãy xem aggregation pipeline như dây chuyền lắp ráp cho dữ liệu của bạn. Mỗi document từ collection đi vào một đầu của pipeline và đi qua nhiều “trạm” (stage) nơi nó được lọc, biến đổi, nhóm hoặc bổ sung dữ liệu. 

Đầu ra của một stage trở thành đầu vào cho stage tiếp theo, cho phép bạn chia nhỏ các phép biến đổi dữ liệu phức tạp thành những bước nhỏ, dễ quản lý.

Các pipeline này dùng cách tiếp cận khai báo; bạn chỉ định bạn muốn gì ở mỗi stage thay vì cách tính toán như thế nào. Cách này làm rõ ý đồ xử lý dữ liệu của bạn và cho phép MongoDB xử lý chi tiết thực thi. Cơ sở dữ liệu có thể áp dụng nhiều tối ưu hóa dựa trên cấu trúc pipeline của bạn.

Thứ tự stage có ý nghĩa trong thiết kế pipeline. Lọc document sớm (trước khi nhóm hoặc tính toán phức tạp) sẽ giảm lượng dữ liệu đi qua pipeline. 

Cách tiếp cận này có thể cải thiện hiệu năng đáng kể khi làm việc với các collection lớn. Một pipeline được cấu trúc tốt sẽ chỉ xử lý dữ liệu cần thiết cho kết quả cuối cùng.

• Luồng document: Document đi vào → biến đổi qua các stage → kết quả xuất ra
• Thiết kế khai báo: Chỉ định điều bạn muốn, không phải cách tính
• Xử lý có thứ tự: Đầu ra của mỗi stage cấp dữ liệu trực tiếp cho stage kế tiếp
• Lợi ích hiệu năng: Lọc sớm giảm khối lượng dữ liệu ở các stage sau

Phân loại stage và các nhóm chức năng

Các stage trong aggregation của MongoDB thuộc bốn nhóm chính theo mục đích. Các stage lọc như $match hoạt động như truy vấn, chọn chỉ những document đáp ứng tiêu chí nhất định. Điều này giúp thu hẹp dữ liệu trước khi thực hiện các thao tác phức tạp hơn.

Các stage định hình lại cấu trúc như $project hoặc $addFields cho phép bạn bao gồm, loại trừ, đổi tên trường, hoặc tạo các trường tính toán dựa trên giá trị hiện có. Các stage này giúp đơn giản hóa document bằng cách chỉ giữ thông tin liên quan và thêm các giá trị tính toán cần cho phân tích.

Khi cần kết hợp nhiều document dựa trên đặc điểm chung, các stage nhóm phát huy tác dụng. Stage $group là “trụ cột”, cho phép tính đếm, tổng, trung bình và các giá trị tổng hợp khác theo nhóm document. Điều này biến hàng nghìn bản ghi riêng lẻ thành các tóm tắt có ý nghĩa trả lời câu hỏi phân tích của bạn.

Để hoàn thiện bức tranh dữ liệu, các stage join như $lookup cho phép bạn kết hợp thông tin từ nhiều collection. Điều này giúp làm giàu document bằng dữ liệu liên quan, tương tự join trong SQL nhưng thích nghi cho mô hình document của MongoDB. Khả năng tham chiếu dữ liệu giữa các collection giúp duy trì bình thường hóa dữ liệu đúng cách trong khi vẫn trả về kết quả đầy đủ trong một thao tác.

• Lọc: Chọn document cụ thể bằng $match theo tiêu chí
• Định hình lại: Biến đổi cấu trúc document với $project và $addFields
• Nhóm: Kết hợp và tóm tắt bằng $group để có các giá trị tổng hợp
• Join: Kết nối dữ liệu liên quan giữa các collection bằng $lookup

Để tham khảo đầy đủ về khả năng và toán tử, bạn có thể xem MongoDB Aggregation Pipeline Manual.

Thiết lập môi trường cho hướng dẫn

Trước khi bắt đầu với aggregation pipeline của MongoDB, bạn cần một môi trường hoạt động với PyMongo và quyền truy cập bộ dữ liệu mẫu. Phần này hướng dẫn bạn quá trình thiết lập với bộ dữ liệu sample_analytics, chứa dữ liệu tài chính rất phù hợp để minh họa các khái niệm aggregation.

Cài đặt PyMongo

PyMongo là driver Python chính thức của MongoDB. Bạn có thể cài bằng pip trên cả macOS và Windows:

# Install PyMongo using pip
pip install pymongo

# Or if you're using conda
conda install -c conda-forge pymongoĐược hỗ trợ bởi

• Tương thích phiên bản: PyMongo 4.x hoạt động với Python 3.7+
• Xử lý phụ thuộc: Trình cài đặt tự động xử lý các phụ thuộc cần thiết
• Hỗ trợ nền tảng: Hoạt động như nhau trên Windows, macOS và Linux

Truy cập bộ dữ liệu mẫu

MongoDB cung cấp các bộ dữ liệu mẫu để bạn dùng mà không cần tự tạo dữ liệu. Cách dễ nhất là dùng MongoDB Atlas (phiên bản đám mây):

1. Tạo tài khoản MongoDB Atlas miễn phí tại mongodb.com/cloud/atlas
2. Thiết lập một cluster gói miễn phí

2.1. Chọn phiên bản miễn phí trọn đời

2.2. Đặt tên cho cluster của bạn

2.3. Nhấp “Deploy”

2.4. Sao chép tên người dùng và mật khẩu của bạn vào nơi an toàn

2.5. Nhấp “Create database user”

2.6. Chọn “Drivers” cho phương thức kết nối của bạn

2.7. Sao chép URI kết nối cho bước tiếp theo

3.1. Nhấp vào ba chấm trong trang xem cluster

3.2. Chọn “Load sample dataset”

3.3. Từ danh sách, chọn sample_analytics và đợi tải xong.

Với cài đặt MongoDB cục bộ, bạn có thể nạp bộ dữ liệu mẫu bằng:

# Download and restore the sample dataset
python -m pip install pymongo[srv]
python -c "from pymongo import MongoClient; MongoClient().admin.command('getParameter', '*')"Được hỗ trợ bởi

• Kích thước dữ liệu: Bộ dữ liệu sample_analytics đủ nhỏ (~10MB) để hoạt động tốt trên gói miễn phí
• Cấu trúc collection: Gồm các collection customers và accounts với dữ liệu quan hệ
• Tính thực tế của dữ liệu: Dựa trên các mẫu tài chính thực tế để có ví dụ aggregation ý nghĩa

Kết nối tới MongoDB

Giờ bạn đã có MongoDB và bộ dữ liệu mẫu sẵn sàng, hãy viết mã để kết nối và kiểm tra thiết lập của chúng ta:

from pymongo.mongo_client import MongoClient
from pymongo.server_api import ServerApi

uri = "mongodb+srv://bex:[email protected]/?retryWrites=true&w=majority&appName=Cluster0"

# Create a new client and connect to the server
client = MongoClient(uri, server_api=ServerApi('1'))

# Send a ping to confirm a successful connection
try:
   client.admin.command('ping')
   print("Pinged your deployment. You successfully connected to MongoDB!")
except Exception as e:
   print(e)
Pinged your deployment. You successfully connected to MongoDB!
# Access the sample_analytics database
db = client.sample_analytics

# Verify connection by counting documents in collections
customer_count = db.customers.count_documents({})
account_count = db.accounts.count_documents({})

print(f"Found {customer_count} customers and {account_count} accounts in sample_analytics")

# Preview one document from each collection
print("\nSample customer document:")
print(db.customers.find_one())

print("\nSample account document:")
print(db.accounts.find_one())Được hỗ trợ bởi

Đầu ra:

Found 500 customers and 1746 accounts in sample_analytics

Sample customer document:
{'_id': ObjectId('5ca4bbcea2dd94ee58162a68'), 'username': 'fmiller', 'name': 'Elizabeth Ray', 'address': '9286 Bethany Glens\nVasqueztown, CO 22939', 'birthdate': datetime.datetime(1977, 3, 2, 2, 20, 31), 'email': '[email protected]', 'active': True, 'accounts': [371138, 324287, 276528, 332179, 422649, 387979], 'tier_and_details': {'0df078f33aa74a2e9696e0520c1a828a': {'tier': 'Bronze', 'id': '0df078f33aa74a2e9696e0520c1a828a', 'active': True, 'benefits': ['sports tickets']}, '699456451cc24f028d2aa99d7534c219': {'tier': 'Bronze', 'benefits': ['24 hour dedicated line', 'concierge services'], 'active': True, 'id': '699456451cc24f028d2aa99d7534c219'}}}

Sample account document:
{'_id': ObjectId('5ca4bbc7a2dd94ee5816238c'), 'account_id': 371138, 'limit': 9000, 'products': ['Derivatives', 'InvestmentStock']}Được hỗ trợ bởi

• Xác minh: Mã xác nhận kết nối và quyền truy cập bộ dữ liệu
• Xem trước document: Hiển thị cấu trúc các document bạn sẽ làm việc

Khi chạy đoạn mã này, bạn sẽ thấy đầu ra hiển thị số lượng document và bản xem trước của cả document khách hàng và tài khoản. Điều này xác nhận môi trường của bạn đã sẵn sàng cho các ví dụ aggregation pipeline mà chúng ta sẽ khám phá ở các phần tiếp theo.

Để tìm hiểu nhập môn về PyMongo và thiết lập MongoDB cục bộ, hãy xem hướng dẫn PyMongo cho người mới bắt đầu.

Tìm hiểu sâu về Aggregation Pipeline của MongoDB

Giờ chúng ta đã xác nhận kết nối và khám phá cấu trúc document, hãy đi sâu vào framework aggregation của MongoDB. Tính năng mạnh mẽ này cho phép chúng ta xử lý và biến đổi dữ liệu trực tiếp trong cơ sở dữ liệu. Ở phần này, chúng ta sẽ khám phá các stage phổ biến nhất với ví dụ thực tiễn trên bộ dữ liệu mẫu.

Lọc dữ liệu với $match

Stage $match lọc document dựa trên tiêu chí chỉ định. Hãy coi đó là cách tập trung vào đúng dữ liệu bạn quan tâm trước khi thực hiện các thao tác phức tạp hơn.

Hãy tìm tất cả tài khoản premium có hạn mức trên $9.000:

pipeline = [
   {"$match": {"limit": {"$gt": 9000}}}
]

premium_accounts = list(db.accounts.aggregate(pipeline))

print(f"Found {len(premium_accounts)} premium accounts")
print(premium_accounts[0])Được hỗ trợ bởi

Pipeline này dùng stage $match với toán tử so sánh $gt (lớn hơn) để lọc tài khoản. Nó hoạt động tương tự phương thức find() nhưng trong ngữ cảnh pipeline. Truy vấn sẽ kiểm tra từng document trong collection accounts và chỉ giữ những document có trường limit lớn hơn 9000.

Đầu ra:

Found 1701 premium accounts
{'_id': ObjectId('5ca4bbc7a2dd94ee5816238d'), 'account_id': 557378, 'limit': 10000, 'products': ['InvestmentStock', 'Commodity', 'Brokerage', 'CurrencyService']}Được hỗ trợ bởi

Nhìn vào kết quả, ta thấy pipeline xác định được 1701 tài khoản premium trong tổng số 1746 tài khoản của cơ sở dữ liệu. Bước lọc này thu hẹp trọng tâm, giúp các phân tích tiếp theo hiệu quả và có mục tiêu hơn. Trong dự án của bạn, bạn có thể dùng kỹ thuật này để tập trung vào người dùng đang hoạt động, các giao dịch trên một mức nhất định, hoặc sản phẩm thuộc một danh mục cụ thể trước khi phân tích sâu hơn các document đó.

Định hình document với $project, $sort, $limit và $skip

Các stage này giúp bạn kiểm soát trường nào được bao gồm và cách sắp xếp kết quả. Dưới đây là ví dụ lấy top 5 tài khoản có hạn mức cao nhất, chỉ hiển thị thông tin thiết yếu:

pipeline = [
   {"$project": {
       "_id": 0,
       "account_id": 1,
       "limit": 1,
       "product_count": {"$size": "$products"}
   }},
   {"$sort": {"limit": -1}},
   {"$limit": 5}
]

top_accounts = list(db.accounts.aggregate(pipeline))
for account in top_accounts:
   print(account)Được hỗ trợ bởi

Pipeline này có ba stage phối hợp với nhau:

1. Stage $project định hình lại mỗi document bằng cách:

• Loại trừ trường _id (đặt về 0)
• Bao gồm các trường account_id và limit (đặt về 1)
• Tạo trường mới tên product_count sử dụng toán tử $size (một trong nhiều toán tử của aggregation pipeline) để đếm số phần tử trong mảng products

2. Stage $sort sắp xếp kết quả theo trường limit giảm dần (-1)

3. Stage $limit chỉ giữ 5 document đầu tiên sau khi sắp xếp

Đầu ra:

{'account_id': 674364, 'limit': 10000, 'product_count': 1}
{'account_id': 278603, 'limit': 10000, 'product_count': 2}
{'account_id': 383777, 'limit': 10000, 'product_count': 5}
{'account_id': 557378, 'limit': 10000, 'product_count': 4}
{'account_id': 198100, 'limit': 10000, 'product_count': 3}Được hỗ trợ bởi

Đầu ra hiển thị năm tài khoản có hạn mức cao nhất, đều ở mức $10.000. Chúng ta cũng thấy số lượng sản phẩm mỗi tài khoản nắm giữ qua trường tính toán product_count. 

Điều này cho phép trình bày dữ liệu gọn gàng, tập trung đúng nhu cầu thay vì trả về mọi trường. Khi xây dựng bảng điều khiển hay báo cáo cho ứng dụng của bạn, bạn có thể dùng kỹ thuật tương tự để chỉ hiển thị thông tin liên quan nhất cho người dùng, giảm truyền dữ liệu và đơn giản hóa giao diện.

Nhóm và tổng hợp với $group

Stage $group là nơi aggregation tỏa sáng. Bạn có thể phân loại document và tính các chỉ số trên mỗi nhóm. Hãy tìm hạn mức trung bình theo loại sản phẩm:

pipeline = [
   {"$unwind": "$products"},  # First unwind the products array
   {"$group": {
       "_id": "$products",
       "avg_limit": {"$avg": "$limit"},
       "count": {"$sum": 1}
   }},
   {"$sort": {"avg_limit": -1}}
]

product_analysis = list(db.accounts.aggregate(pipeline))
for product in product_analysis:
   print(f"Product: {product['_id']}")
   print(f"  Average limit: ${product['avg_limit']:.2f}")
   print(f"  Number of accounts: {product['count']}")Được hỗ trợ bởi

Pipeline này dùng nhiều toán tử để phân tích dữ liệu sản phẩm:

1. Toán tử $unwind tách các document có nhiều sản phẩm thành các document riêng. Ví dụ, một tài khoản với ["Derivatives", "InvestmentStock"] sẽ thành hai document, mỗi document cho một sản phẩm.
2. Stage $group sau đó:

• Nhóm theo tên sản phẩm (trường _id trong $group quyết định khóa nhóm)
• Tính hạn mức trung bình cho mỗi sản phẩm bằng bộ tích lũy $avg
• Đếm số tài khoản cho mỗi sản phẩm dùng $sum: 1 (cộng 1 cho mỗi document)

3. Stage $sort sắp xếp kết quả theo hạn mức trung bình giảm dần

Đầu ra:

Product: Commodity
 Average limit: $9963.89
 Number of accounts: 720
Product: Brokerage
 Average limit: $9960.86
 Number of accounts: 741
Product: InvestmentStock
 Average limit: $9955.90
 Number of accounts: 1746
Product: InvestmentFund
 Average limit: $9951.92
 Number of accounts: 728
Product: Derivatives
 Average limit: $9951.84
 Number of accounts: 706
Product: CurrencyService
 Average limit: $9946.09
 Number of accounts: 742Được hỗ trợ bởi

Kết quả cho thấy sản phẩm Commodity gắn với hạn mức tài khoản trung bình cao nhất, trong khi CurrencyService có mức thấp nhất. 

Dạng phân tích này giúp xác định tương quan giữa danh mục sản phẩm và khả năng chi tiêu của khách hàng. 

Trong ứng dụng của bạn, bạn có thể dùng kỹ thuật tương tự để phân tích doanh số theo danh mục, mức độ tương tác người dùng theo tính năng, hoặc lỗi theo module — bất kỳ kịch bản nào cần tóm tắt dữ liệu theo nhóm thay vì xem xét từng bản ghi riêng lẻ.

Join các collection với $lookup và $unwind

Khi dữ liệu của bạn nằm ở nhiều collection, $lookup giúp gom chúng lại. Hãy tìm khách hàng cùng các tài khoản liên quan:

pipeline = [
   {"$match": {"username": "fmiller"}},  # Find a specific customer
   {"$lookup": {
       "from": "accounts",  # Collection to join with
       "localField": "accounts",  # Field from customers collection
       "foreignField": "account_id",  # Field from accounts collection
       "as": "account_details"  # Name for the new array field
   }},
   {"$project": {
       "name": 1,
       "accounts": 1,
       "account_details.account_id": 1,
       "account_details.limit": 1,
       "account_details.products": 1
   }}
]

customer_accounts = list(db.customers.aggregate(pipeline))

print(f"Customer: {customer_accounts[0]['name']}")
print(f"Has {len(customer_accounts[0]['account_details'])} accounts:")

for account in customer_accounts[0]['account_details']:
   print(f"  Account {account['account_id']}: ${account['limit']} limit with products: {', '.join(account['products'])}")Được hỗ trợ bởi

Pipeline này thể hiện cách join dữ liệu liên quan giữa các collection:

1. Stage $match tìm một khách hàng cụ thể theo username
2. Stage $lookup thực hiện left outer join với collection accounts:

• from: chỉ định collection để join
• localField: trường trong collection hiện tại (customers) để đối sánh
• foreignField: trường trong collection mục tiêu (accounts) để đối sánh
• as: tên trường mảng mới sẽ chứa các document khớp

3. Stage $project định hình đầu ra để chỉ hiển thị các trường liên quan

Đầu ra:

Customer: Elizabeth Ray
Has 6 accounts:
 Account 371138: $9000 limit with products: Derivatives, InvestmentStock
 Account 324287: $10000 limit with products: Commodity, CurrencyService, Derivatives, InvestmentStock
 Account 276528: $10000 limit with products: InvestmentFund, InvestmentStock
 Account 332179: $10000 limit with products: Commodity, CurrencyService, InvestmentFund, Brokerage, InvestmentStock
 Account 422649: $10000 limit with products: CurrencyService, InvestmentStock
 Account 387979: $10000 limit with products: Brokerage, Derivatives, InvestmentFund, Commodity, InvestmentStockĐược hỗ trợ bởi

Đầu ra cung cấp cái nhìn toàn diện về danh mục tài chính của Elizabeth Ray, hiển thị cả sáu tài khoản và các sản phẩm liên quan chỉ trong một kết quả truy vấn. 

Điều này cho phép bạn có góc nhìn 360 độ về các mối quan hệ dữ liệu mà không cần nhiều truy vấn hay join phía client. 

Trong ứng dụng của bạn, hãy cân nhắc nơi bạn đã tách dữ liệu ra nhiều collection để bình thường hóa nhưng cần lắp ghép lại cho mục đích phân tích hay hiển thị. Ví dụ phổ biến gồm hồ sơ người dùng với lịch sử hoạt động, sản phẩm với trạng thái tồn kho, hoặc nội dung với bình luận liên quan.

Kết hợp nhiều stage cho phân tích phức tạp

Giờ hãy giải một câu hỏi kinh doanh phức tạp hơn: “Hạn mức tài khoản trung bình của khách hàng là bao nhiêu, nhóm theo hạng dịch vụ của họ?” Điều này yêu cầu xử lý một đối tượng lồng trong document khách hàng (tier_and_details) rồi liên kết thông tin này với collection accounts.

Trước tiên, để hiểu cấu trúc trong collection customers, hãy kiểm tra trường tier_and_details và trường accounts của một khách hàng mẫu. Trường tier_and_details là một đối tượng nơi mỗi khóa là định danh cho một gói hạng, và giá trị chứa chi tiết như tên hạng. Trường accounts là mảng ID tài khoản gắn với khách hàng đó.

# First, let's look at the structure of 'tier_and_details' again
sample_customer = db.customers.find_one({"username": "fmiller"}) # Using a known customer for consistency
print("Tier structure example for customer 'fmiller':")
print(sample_customer['tier_and_details'])
print(f"Customer 'fmiller' has account IDs: {sample_customer['accounts']}")Được hỗ trợ bởi

Đầu ra:

Tier structure example for customer 'fmiller':
{'0df078f33aa74a2e9696e0520c1a828a': {'tier': 'Bronze', 'id': '0df078f33aa74a2e9696e0520c1a828a', 'active': True, 'benefits': ['sports tickets']}, '699456451cc24f028d2aa99d7534c219': {'tier': 'Bronze', 'benefits': ['24 hour dedicated line', 'concierge services'], 'active': True, 'id': '699456451cc24f028d2aa99d7534c219'}}
Customer 'fmiller' has account IDs: [371138, 324287, 276528, 332179, 422649, 387979]Được hỗ trợ bởi

pipeline = [
   # Step 1: Project necessary fields, including 'accounts' and convert 'tier_and_details'
   {"$project": {
       "tiers_array": {"$objectToArray": "$tier_and_details"}, # Convert object to array
       "customer_account_ids": "$accounts",  # Explicitly carry over the customer's account IDs
       "_id": 1  # Keep customer _id for later
   }},
   # Step 2: Unwind the new 'tiers_array' to process each tier object separately
   {"$unwind": "$tiers_array"},
   # Step 3: Reshape to clearly define the tier and keep customer account IDs
   {"$project": {
       "tier": "$tiers_array.v.tier",  # Extract the tier name
       "customer_account_ids": 1,  # Ensure account IDs are still present
       "customer_id": "$_id"  # Rename _id to customer_id for clarity
   }},
   # Step 4: Look up account details using the customer_account_ids
   {"$lookup": {
       "from": "accounts",  # Target collection
       "localField": "customer_account_ids",  # Array of account IDs from the customer
       "foreignField": "account_id",  # Field in the 'accounts' collection
       "as": "matched_account_details"  # New array with joined account documents
   }},
   # Step 5: Unwind the 'matched_account_details' array.
   {"$unwind": "$matched_account_details"},
   # Step 6: Group by tier to calculate statistics
   {"$group": {
       "_id": "$tier",  # Group by the tier name
       "avg_limit": {"$avg": "$matched_account_details.limit"}, # Calculate average limit
       "total_accounts_in_tier": {"$sum": 1},  # Count how many accounts fall into this tier
       "unique_customers_in_tier": {"$addToSet": "$customer_id"}  # Count unique customers in this tier
   }},
   # Step 7: Format the final output
   {"$project": {
       "tier_name": "$_id",  # Rename _id to tier_name
       "average_account_limit": "$avg_limit",
       "number_of_accounts": "$total_accounts_in_tier",
       "number_of_customers": {"$size": "$unique_customers_in_tier"},  # Get the count of unique customers
       "_id": 0  # Exclude the default _id
   }},
   # Step 8: Sort by average limit
   {"$sort": {"average_account_limit": -1}}
]

tier_analysis = list(db.customers.aggregate(pipeline))

print("\nTier Analysis Results:")

for tier_data in tier_analysis:
   print(f"Tier: {tier_data['tier_name']}")
   print(f"  Average Account Limit: ${tier_data['average_account_limit']:.2f}")
   print(f"  Number of Accounts in this Tier: {tier_data['number_of_accounts']}")
   print(f"  Number of Unique Customers in this Tier: {tier_data['number_of_customers']}")Được hỗ trợ bởi

Pipeline này chia nhỏ phân tích phức tạp thành tám bước dễ quản lý:

1. $project: Chúng ta bắt đầu bằng cách biến đổi đối tượng tier_and_details. Toán tử $objectToArray chuyển đối tượng này thành mảng cặp khóa-giá trị (tiers_array). Điều này rất quan trọng vì các stage như $unwind hoạt động trên mảng. Quan trọng là chúng ta cũng truyền rõ ràng mảng accounts từ document khách hàng thành customer_account_ids và _id của khách hàng.
2. $unwind: Stage này phá vỡ tiers_array, tạo một document riêng cho mỗi mục hạng mà khách hàng có. Mỗi document mới vẫn chứa customer_account_ids và _id gốc của khách hàng.
3. $project: Chúng ta định hình lại document để trích xuất rõ tên hạng (ví dụ, "Bronze") từ cấu trúc lồng ($tiers_array.v.tier) và đổi tên _id của khách hàng thành customer_id cho rõ ràng. customer_account_ids được truyền qua.
4. $lookup: Đây là nơi chúng ta join với collection accounts. Chúng ta dùng customer_account_ids (mảng số tài khoản từ document khách hàng) làm localField. foreignField là account_id của collection accounts. MongoDB sẽ tìm mọi tài khoản có account_id xuất hiện trong mảng customer_account_ids, thêm chúng như một mảng vào trường matched_account_details.
5. $unwind: Chúng ta unwind matched_account_details. Giờ đây, nếu một cặp khách hàng-hạng liên kết với nhiều tài khoản, ta có một document riêng cho từng tài khoản cụ thể gắn với khách hàng và hạng đó.
6. $group: Chúng ta nhóm các document theo hạng. Với mỗi hạng, ta tính avg_limit dùng $avg trên limit từ chi tiết tài khoản đã join. Ta đếm total_accounts_in_tier bằng $sum: 1. Ta cũng dùng $addToSet với customer_id để thu thập ID khách hàng duy nhất thuộc mỗi hạng.
7. $project: Định hình đầu ra cuối. Đổi tên _id (tên hạng từ stage group) thành tier_name. Dùng $size để lấy số lượng unique_customers_in_tier.
8. $sort: Sắp xếp kết quả theo average_account_limit` giảm dần, để hạng có hạn mức trung bình cao nhất xuất hiện trước.

Tier Analysis Results:
Tier: Silver
 Average Account Limit: $9974.55
 Number of Accounts in this Tier: 393
 Number of Unique Customers in this Tier: 95
Tier: Bronze
 Average Account Limit: $9964.11
 Number of Accounts in this Tier: 418
 Number of Unique Customers in this Tier: 93
Tier: Platinum
 Average Account Limit: $9962.53
 Number of Accounts in this Tier: 427
 Number of Unique Customers in this Tier: 101
Tier: Gold
 Average Account Limit: $9962.44
 Number of Accounts in this Tier: 426
 Number of Unique Customers in this Tier: 99Được hỗ trợ bởi

Kết quả cho thấy một thứ bậc hơi khác: khách hàng hạng Silver trung bình có tài khoản với hạn mức cao nhất, tiếp theo là Bronze, rồi Platinum, và cuối cùng là Gold.

Dạng insight này hữu ích để hiểu các phân khúc khách hàng. Ví dụ, một tổ chức tài chính có thể dùng thông tin này để tìm hiểu vì sao các hạng Silver và Bronze có hạn mức trung bình cao, hoặc để điều chỉnh chiến dịch marketing hay cung cấp dịch vụ cao cấp. 

Việc dùng $objectToArray là chìa khóa để mở dữ liệu hạng lồng nhau, và việc truyền cẩn thận customer_account_ids đảm bảo $lookup có thể liên kết khách hàng với các tài khoản cụ thể của họ.

Khi gặp các đối tượng lồng nhau trong bộ dữ liệu của bạn cần dùng trong aggregation (như user preferences, product attributes hoặc configuration settings), hãy nhớ kỹ thuật $objectToArray. 

Luôn đảm bảo mọi trường cần cho các stage sau, đặc biệt cho thao tác $lookup, được đưa vào rõ ràng trong các stage $project của bạn. Cách tiếp cận có cấu trúc để chia nhỏ dữ liệu phức tạp giúp rút ra insight có ý nghĩa trực tiếp trong MongoDB.

Các mẫu pipeline nâng cao

Ngoài các biến đổi tuần tự, framework aggregation của MongoDB cung cấp các mẫu tinh vi để giải quyết truy vấn phân tích phức tạp. Khi bạn cần phân tích dữ liệu từ nhiều góc độ đồng thời hoặc thực hiện tính toán dựa trên một tập document cuộn, các mẫu nâng cao này mang đến giải pháp mạnh mẽ ngay trong cơ sở dữ liệu.

Phân tích đa chiều với $facet

Stage $facet cho phép bạn thực thi nhiều sub-pipeline aggregation trong một stage, sử dụng cùng tập document đầu vào. Hãy tưởng tượng bạn cần phân loại sản phẩm theo khoảng giá và đồng thời liệt kê 5 thương hiệu sản phẩm phổ biến nhất từ cùng một collection sản phẩm. $facet xử lý các góc nhìn phân tích khác nhau này song song.

Mỗi sub-pipeline trong $facet hoạt động độc lập trên các document đầu vào và tạo ra mảng document đầu ra riêng. Điều này có nghĩa bạn có thể thu thập nhiều chỉ số hoặc tóm tắt — như những gì cần cho một dashboard toàn diện — tất cả trong một truy vấn cơ sở dữ liệu. 

Ví dụ, bạn có thể lấy số lượng người dùng đang hoạt động, phân tách người dùng theo hạng đăng ký, và danh sách người dùng mới tham gia gần đây, tất cả đồng thời từ cùng bộ dữ liệu người dùng. Điều này cho phép tạo các báo cáo phong phú, đa chiều mà không phải chịu chi phí của nhiều lần gọi cơ sở dữ liệu, giúp tối ưu truy xuất dữ liệu cho các giao diện phức tạp.

Phân tích theo thời gian với window function

Được giới thiệu từ MongoDB 5.0, window function thực hiện tính toán trên một tập document liên quan đến document hiện tại, gọi là “cửa sổ”. 

Điều này hữu ích cho dữ liệu chuỗi thời gian hoặc bất kỳ bộ dữ liệu có thứ tự nào mà ngữ cảnh từ các document lân cận là quan trọng. Ví dụ, bạn có thể muốn tính trung bình động 7 ngày của doanh số hoặc tìm tổng tích lũy giao dịch cho mỗi khách hàng theo thời gian.

Window function thường được dùng trong stage $setWindowFields. Stage này cho phép bạn định nghĩa partition (nhóm document, như doanh số theo sản phẩm) và thứ tự sắp xếp trong mỗi partition (như theo ngày). 

Sau đó, bạn có thể áp dụng window function như $avg, $sum, $min, $max, hoặc các hàm chuyên biệt như $derivative hay $integral trên một cửa sổ xác định (ví dụ, 3 document trước đó và document hiện tại).

Hãy xem xét việc tính tổng lũy kế doanh số sản phẩm. Một pipeline dùng $setWindowFields có thể như sau:

pipeline = [
   {"$match": {"category": "Electronics"}}, # Filter for electronics
   {"$sort": {"sale_date": 1}}, # Sort by sale date
   {"$setWindowFields": {
       "partitionBy": "$product_id", # Calculate running total per product
       "sortBy": {"sale_date": 1},
       "output": {
           "running_total_sales": {
               "$sum": "$sale_amount",
               "window": {
                   "documents": ["unbounded", "current"] # Sum from start to current document
               }
           }
       }
   }}
]
# electronics_sales_with_running_total = list(db.sales.aggregate(pipeline))Được hỗ trợ bởi

Trong pipeline mang tính khái niệm này, với mỗi giao dịch điện tử, một trường running_total_sales được thêm vào. Trường này biểu diễn tổng sale_amount cho sản phẩm đó từ giao dịch sớm nhất đến giao dịch hiện tại. 

Những tính toán như vậy, trước đây cần logic phức tạp phía client hoặc nhiều truy vấn, nay có thể thực hiện trực tiếp trong cơ sở dữ liệu, đơn giản hóa mã ứng dụng và cải thiện hiệu năng cho phân tích xu hướng hoặc so sánh theo giai đoạn. 

Kết luận

Aggregation pipeline của MongoDB cung cấp phương pháp có cấu trúc để xử lý dữ liệu ngay trong cơ sở dữ liệu của bạn. Bằng cách liên kết các stage thao tác riêng biệt, bạn có thể thực hiện các biến đổi dữ liệu phức tạp, như lọc bản ghi, định hình document, nhóm thông tin và kết hợp dữ liệu từ nhiều collection. 

Cách tiếp cận này giúp tinh gọn quy trình dữ liệu, cho phép trích xuất ý nghĩa nhanh hơn từ các bộ dữ liệu lớn, như minh họa qua các ví dụ PyMongo. Xử lý dữ liệu gần nguồn thường mang lại hiệu năng tốt hơn cho truy vấn phân tích.

Thành thạo aggregation pipeline chuẩn bị cho bạn đối mặt với nhiều thách thức phân tích dữ liệu. Để áp dụng những gì đã học, bạn có thể thử một dự án thực hành như Xây dựng Retail Data Pipeline. 

Để có góc nhìn rộng hơn về hệ thống dữ liệu và cách kỹ năng MongoDB phù hợp với bức tranh tổng thể, lộ trình Data Engineer with Python cung cấp nội dung học sâu. Nếu bạn hướng tới thăng tiến nghề nghiệp, xem qua Câu hỏi phỏng vấn MongoDB cũng hữu ích để hiểu kỳ vọng giải quyết vấn đề điển hình.