Python में PyMongo के साथ MongoDB Aggregation Pipeline ट्यूटोरियल

बड़े डेटासेट्स के साथ काम करते समय प्रदर्शन बनाए रखते हुए सार्थक पैटर्न और इनसाइट्स निकालना एक चुनौती होता है। जब आपके अनुप्रयोग MongoDB में डेटा स्टोर करते हैं, तो जटिल क्वेरी और रूपांतरण सीधे डेटाबेस में चलाना, डेटा को बाहरी विश्लेषण टूल्स में ले जाने की तुलना में अधिक तेज़ हो सकता है। MongoDB aggregation pipelines आपको डेटा को वहीं प्रोसेस, ट्रांसफॉर्म और विश्लेषित करने देती हैं जहाँ वह रहता है—इस प्रकार एक प्रभावी समाधान प्रदान करती हैं।

आप सरल ऑपरेशंस को क्रम में जोड़कर कस्टम डेटा प्रोसेसिंग वर्कफ़्लो बना सकते हैं। पाइपलाइन में प्रत्येक स्टेज दस्तावेज़ों को ट्रांसफॉर्म करता है और परिणामों को अगले स्टेज तक पहुँचाता है। उदाहरण के लिए, आपको दिनांक-सीमा के आधार पर रिकॉर्ड्स फ़िल्टर करने, श्रेणी के अनुसार समूहित करने, सांख्यिकीय माप निकालने, और आउटपुट को फ़ॉर्मेट करने की आवश्यकता हो सकती है—यह सब एकल डेटाबेस ऑपरेशन के माध्यम से संभव है जो डेटा को उसके स्रोत के पास प्रोसेस करता है।

इस लेख में, आप सामान्य डेटा चुनौतियों को हल करने के लिए aggregation pipelines बनाना सीखेंगे, और सभी उदाहरण PyMongo (MongoDB का आधिकारिक Python क्लाइंट) में दिखाए गए हैं।

हालाँकि यह लेख aggregation pipelines पर केंद्रित है, यदि आप Python के साथ MongoDB का उपयोग करने में नए हैं, तो Introduction to MongoDB in Python कोर्स एक विस्तृत शुरुआती बिंदु प्रदान करता है। इस लेख से आपको इतनी बुनियाद मिल जाएगी कि आप इन aggregation अवधारणाओं को MongoDB की क्वेरी भाषा में स्वयं या भाषा मॉडल्स की सहायता से अनुवादित कर सकें।

MongoDB में Aggregation Pipelines क्या हैं?

मान लीजिए आपको कई प्रोडक्ट्स के ग्राहक रिव्यू का विश्लेषण कर संतुष्टि के रुझान समझने हैं। पारंपरिक क्वेरियाँ डेटा तो ले आएँगी, लेकिन वे उस जानकारी को जोड़ने, विश्लेषित करने और उपयोगी सारांशों में बदलने में मदद नहीं करतीं।

MongoDB की aggregation pipelines इसे हल करती हैं क्योंकि वे एक-दूसरे पर आधारित ऑपरेशंस की श्रृंखला के माध्यम से डेटा प्रोसेस करने का संरचित तरीका प्रदान करती हैं।

पाइपलाइन आर्किटेक्चर और डॉक्यूमेंट फ्लो

Aggregation pipelines को अपने डेटा के लिए असेंबली लाइन समझें। कलेक्शन का प्रत्येक डॉक्यूमेंट पाइपलाइन के एक सिरे से प्रवेश करता है और विभिन्न स्टेशनों (स्टेजेज़) से होकर गुजरता है जहाँ उसे फ़िल्टर, ट्रांसफॉर्म, ग्रुप या एन्हांस किया जाता है।

एक स्टेज का आउटपुट अगले स्टेज का इनपुट बनता है, जिससे आप जटिल डेटा ट्रांसफ़ॉर्मेशन को छोटे, संभालने योग्य चरणों में बाँट सकते हैं।

ये pipelines घोषणात्मक (declarative) दृष्टिकोण अपनाती हैं; आप प्रत्येक स्टेज पर क्या चाहिए यह बताते हैं, न कि इसे कैसे कंप्यूट करना है। यह तरीका आपके इरादों को स्पष्ट करता है और MongoDB को निष्पादन के विवरण सँभालने देता है। तब डेटाबेस आपकी पाइपलाइन संरचना के आधार पर विभिन्न अनुकूलन लागू कर सकता है।

पाइपलाइन डिज़ाइन में स्टेज के क्रम का महत्व होता है। शुरुआती चरणों में फ़िल्टरिंग (ग्रुपिंग या जटिल कैलकुलेशंस से पहले) पाइपलाइन से गुज़रने वाले डेटा की मात्रा कम कर देती है।

यह तरीका बड़े कलेक्शंस पर काम करते समय प्रदर्शन में उल्लेखनीय सुधार ला सकता है। एक सुव्यवस्थित पाइपलाइन केवल उन्हीं डेटा को प्रोसेस करती है जिसकी आपके अंतिम परिणामों के लिए आवश्यकता है।

डॉक्यूमेंट फ्लो: डॉक्यूमेंट प्रवेश करते हैं → स्टेजेज़ से ट्रांसफॉर्म होते हैं → परिणाम निकलते हैं
घोषणात्मक डिज़ाइन: क्या चाहिए यह बताइए, कैसे निकालना है यह नहीं
क्रमबद्ध प्रोसेसिंग: हर स्टेज का आउटपुट सीधे अगले स्टेज में जाता है
प्रदर्शन लाभ: शुरुआती फ़िल्टरिंग बाद के स्टेजेज़ में डेटा वॉल्यूम घटाती है

स्टेज की प्रकारिकी और कार्यात्मक श्रेणियाँ

MongoDB के aggregation स्टेज उनके उद्देश्य के आधार पर चार मुख्य श्रेणियों में आते हैं। $match जैसे फ़िल्टरिंग स्टेज क्वेरियों की तरह काम करते हैं और केवल वे डॉक्यूमेंट चुनते हैं जो विशिष्ट मानदंडों को पूरा करते हैं। इससे जटिल ऑपरेशंस से पहले आपका डेटासेट संकरा होता है।

रिशेपिंग स्टेज डॉक्यूमेंट संरचना को बदलते हैं। $project या $addFields का उपयोग करके आप फ़ील्ड शामिल/बहिष्कृत/रीनेम कर सकते हैं, या मौजूदा मानों के आधार पर गणना किए गए फ़ील्ड बना सकते हैं। ये स्टेज प्रासंगिक जानकारी रखकर और विश्लेषण हेतु आवश्यक कंप्यूटेड वैल्यू जोड़कर डॉक्यूमेंट्स को सरल बनाते हैं।

जब आपको साझा विशेषताओं के आधार पर कई डॉक्यूमेंट्स को संयोजित करना होता है, तो ग्रुपिंग स्टेज काम आते हैं। यहाँ $group मुख्य भूमिका निभाता है—यह डॉक्यूमेंट समूहों पर काउंट, सम, औसत और अन्य समेकित मान निकालने देता है। यह हज़ारों व्यक्तिगत रिकॉर्ड्स को अर्थपूर्ण सारांशों में बदल देता है जो आपके विश्लेषणात्मक प्रश्नों का उत्तर देते हैं।

आपके डेटा को पूर्ण बनाने के लिए, $lookup जैसे जॉइन स्टेज कई कलेक्शंस की जानकारी को एक साथ लाने देते हैं। यह SQL joins जैसा है, पर MongoDB के डॉक्यूमेंट मॉडल के अनुरूप। कलेक्शंस के आर-पार डेटा संदर्भित करने की क्षमता उपयुक्त नॉर्मलाइज़ेशन बनाए रखते हुए एक ही ऑपरेशन में पूर्ण परिणाम देने में मदद करती है।

फ़िल्टरिंग: मानदंड के आधार पर $match से विशिष्ट डॉक्यूमेंट चुनें
रिशेपिंग: $project और $addFields से डॉक्यूमेंट संरचना बदलें
ग्रुपिंग: समेकित मानों हेतु $group के साथ संयोजित और सारांशित करें
जॉइनिंग: $lookup से कलेक्शंस के बीच संबंधित डेटा जोड़ें

क्षमताओं और ऑपरेटर्स की पूर्ण संदर्भ सूची के लिए आप MongoDB Aggregation Pipeline Manual देख सकते हैं।

ट्यूटोरियल के लिए वातावरण तैयार करना

MongoDB aggregation pipelines में उतरने से पहले, आपको PyMongo और एक सैंपल डेटासेट के साथ कार्यशील वातावरण चाहिए। यह अनुभाग sample_analytics डेटासेट के साथ सेटअप प्रक्रिया बताता है, जिसमें वित्तीय डेटा है—aggregation अवधारणाएँ दिखाने के लिए बिल्कुल उपयुक्त।

PyMongo इंस्टॉल करना

PyMongo MongoDB का आधिकारिक Python ड्राइवर है। आप इसे macOS और Windows दोनों पर pip से इंस्टॉल कर सकते हैं:

# Install PyMongo using pip
pip install pymongo

# Or if you're using conda
conda install -c conda-forge pymongoसंचालित द्वारा

संस्करण अनुकूलता: PyMongo 4.x, Python 3.7+ के साथ काम करता है
डिपेंडेंसी हैंडलिंग: इंस्टॉलर आवश्यक निर्भरताएँ स्वतः संभालता है
प्लेटफ़ॉर्म सपोर्ट: Windows, macOS और Linux पर समान रूप से काम करता है

सैंपल डेटासेट तक पहुँच

MongoDB ऐसे सैंपल डेटासेट उपलब्ध कराता है जिन्हें आप अपना डेटा बनाए बिना उपयोग कर सकते हैं। सबसे आसान तरीका MongoDB Atlas (क्लाउड संस्करण) का उपयोग करना है:

mongodb.com/cloud/atlas पर एक निःशुल्क MongoDB Atlas अकाउंट बनाएँ
एक फ्री टियर क्लस्टर सेट करें

2.1. फ्री फ़ॉरएवर संस्करण चुनें

2.2. अपने क्लस्टर का नाम दें

2.3. “Deploy” पर क्लिक करें

2.4. अपना यूज़रनेम और पासवर्ड कहीं सुरक्षित कॉपी कर लें

2.5. “Create database user” पर क्लिक करें

2.6. कनेक्शन मेथड के लिए “Drivers” चुनें

2.7. अगले चरण के लिए कनेक्शन URI कॉपी करें

3.1. अपने क्लस्टर्स व्यू में तीन बिंदुओं पर क्लिक करें

3.2. “Load sample dataset” चुनें

3.3. सूची में से sample_analytics चुनें और लोड होने दें।

लोकल MongoDB इंस्टॉलेशन के लिए, आप निम्न का उपयोग कर सैंपल डेटासेट लोड कर सकते हैं:

# Download and restore the sample dataset
python -m pip install pymongo[srv]
python -c "from pymongo import MongoClient; MongoClient().admin.command('getParameter', '*')"संचालित द्वारा

डेटा आकार: sample_analytics डेटासेट इतना छोटा है (~10MB) कि फ्री टियर पर भी अच्छी तरह काम करता है
कलेक्शन संरचना: इसमें रिलेशनशिप डेटा के साथ customers और accounts कलेक्शंस हैं
डेटा यथार्थवाद: सार्थक aggregation उदाहरणों के लिए यथार्थवादी वित्तीय पैटर्न पर आधारित

MongoDB से कनेक्ट करना

अब जबकि MongoDB और सैंपल डेटासेट तैयार हैं, आइए कनेक्ट करने और सेटअप सत्यापित करने के लिए कोड लिखें:

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())संचालित द्वारा

आउटपुट:

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']}संचालित द्वारा

सत्यापन: कोड आपके कनेक्शन और डेटासेट एक्सेस की पुष्टि करता है
डॉक्यूमेंट पूर्वावलोकन: उन डॉक्यूमेंट्स की संरचना दिखाता है जिनके साथ आप काम करेंगे

जब आप यह कोड चलाते हैं, तो आपको डॉक्यूमेंट काउंट्स और ग्राहक व खाता डॉक्यूमेंट्स का पूर्वावलोकन दिखाई देना चाहिए। यह पुष्टि करता है कि हमारा वातावरण अगले सेक्शंस में देखने वाले aggregation pipeline उदाहरणों के लिए तैयार है।

PyMongo और लोकल MongoDB सेटअप के परिचय के लिए हमारा शुरुआती PyMongo ट्यूटोरियल देखें।

MongoDB Aggregation Pipelines का गहन विवरण

अब जबकि हमने अपना कनेक्शन सत्यापित कर लिया है और डॉक्यूमेंट संरचना देख ली है, आइए MongoDB के aggregation फ़्रेमवर्क में गहराई से जाएँ। यह शक्तिशाली सुविधा हमें सीधे डेटाबेस में डेटा प्रोसेस और ट्रांसफॉर्म करने देती है। इस अनुभाग में, हम अपने सैंपल डेटासेट का उपयोग करते हुए सबसे सामान्य पाइपलाइन स्टेजेज़ को व्यावहारिक उदाहरणों के साथ देखेंगे।

$match के साथ डेटा फ़िल्टर करना

$match स्टेज निर्दिष्ट मानदंडों के आधार पर डॉक्यूमेंट्स को फ़िल्टर करता है। इसे उन डाटा पर ध्यान केंद्रित करने के तरीके के रूप में समझें जो वास्तव में आपके लिए मायने रखते हैं—जटिल ऑपरेशंस से पहले।

आइए सभी प्रीमियम अकाउंट्स ढूँढें जिनकी लिमिट $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])संचालित द्वारा

यह पाइपलाइन $match स्टेज को तुलना ऑपरेटर $gt (greater than) के साथ इस्तेमाल करती है ताकि अकाउंट्स को फ़िल्टर किया जा सके। यह find() मेथड की तरह ही काम करता है, लेकिन पाइपलाइन के संदर्भ में। क्वेरी accounts कलेक्शन के प्रत्येक डॉक्यूमेंट की जाँच करती है और केवल वही रखती है जिनमें limit फ़ील्ड 9000 से अधिक है।

आउटपुट:

Found 1701 premium accounts
{'_id': ObjectId('5ca4bbc7a2dd94ee5816238d'), 'account_id': 557378, 'limit': 10000, 'products': ['InvestmentStock', 'Commodity', 'Brokerage', 'CurrencyService']}संचालित द्वारा

परिणामों से दिखता है कि पाइपलाइन ने हमारे डेटाबेस के कुल 1746 अकाउंट्स में से 1701 प्रीमियम अकाउंट्स पहचाने। यह फ़िल्टरिंग स्टेप हमारे फोकस को संकरा करता है, जिससे बाद का विश्लेषण अधिक कुशल और लक्षित बनता है। अपनी परियोजनाओं में, आप इस तकनीक का उपयोग सक्रिय उपयोगकर्ताओं, किसी निश्चित राशि से ऊपर के लेनदेन, या किसी विशिष्ट श्रेणी के उत्पादों पर गहराई से विश्लेषण करने से पहले फोकस करने के लिए कर सकते हैं।

$project, $sort, $limit, और $skip से डॉक्यूमेंट्स का रिशेपिंग

ये स्टेज यह नियंत्रित करने में मदद करते हैं कि किन फ़ील्ड्स को शामिल करना है और अपने परिणामों को कैसे व्यवस्थित करना है। यहाँ एक उदाहरण है जो सबसे अधिक लिमिट वाले शीर्ष 5 अकाउंट्स लाता है और केवल आवश्यक जानकारी दिखाता है:

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)संचालित द्वारा

यह पाइपलाइन तीन स्टेजेज़ को साथ में काम कराती है:

1. $project स्टेज प्रत्येक डॉक्यूमेंट को इस प्रकार रिशेप करता है:

_id फ़ील्ड को बहिष्कृत करना (इसे 0 सेट करना)
account_id और limit फ़ील्ड्स को शामिल करना (इन्हें 1 सेट करना)
एक नया फ़ील्ड product_count बनाना जो $size ऑपरेटर (यहाँ कई aggregation pipeline ऑपरेटर्स में से एक) का उपयोग कर products एरे में आइटम्स गिनता है

2. $sort स्टेज परिणामों को limit फ़ील्ड के अनुसार अवरोही क्रम (-1) में सजाता है

3. $limit स्टेज सॉर्टिंग के बाद केवल पहले 5 डॉक्यूमेंट्स को रखता है

आउटपुट:

{'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}संचालित द्वारा

आउटपुट उन पाँच अकाउंट्स को दिखाता है जिनकी लिमिट सबसे अधिक है—सभी $10,000 पर। हम अपने कैलकुलेटेड product_count फ़ील्ड के माध्यम से देख सकते हैं कि प्रत्येक अकाउंट में कितने प्रोडक्ट्स हैं।

यह बिल्कुल वही दिखाने पर केंद्रित एक साफ-सुथरी डेटा प्रस्तुति सक्षम करता है जिसकी ज़रूरत है—सभी फ़ील्ड्स लौटाने के बजाय। जब आप अपनी एप्लिकेशंस के लिए डैशबोर्ड या रिपोर्ट बनाते हैं, तो इसी तरह की तकनीकों से केवल सबसे प्रासंगिक जानकारी प्रस्तुत कर सकते हैं, डेटा ट्रांसफ़र घटा सकते हैं और UI सरल बना सकते हैं।

$group के साथ ग्रुपिंग और एग्रीगेशन

$group स्टेज aggregation की असली ताकत है। आप डॉक्यूमेंट्स को वर्गीकृत कर सकते हैं और प्रत्येक ग्रुप पर मेट्रिक्स की गणना कर सकते हैं। आइए प्रोडक्ट टाइप के अनुसार औसत अकाउंट लिमिट ढूँढें:

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']}")संचालित द्वारा

यह पाइपलाइन कई ऑपरेटर्स का उपयोग करके प्रोडक्ट डेटा का विश्लेषण करती है:

$unwind ऑपरेटर कई प्रोडक्ट्स वाले डॉक्यूमेंट्स को अलग-अलग डॉक्यूमेंट्स में बाँट देता है। उदाहरण के लिए, ["Derivatives", "InvestmentStock"] वाले अकाउंट से दो डॉक्यूमेंट बनते हैं—हर प्रोडक्ट के लिए एक।
$group स्टेज तब:

प्रोडक्ट नाम के आधार पर ग्रुप करता है ($group में _id फ़ील्ड ग्रुपिंग की कुंजी तय करता है)
प्रत्येक प्रोडक्ट के लिए औसत लिमिट $avg accumulator से निकालता है
प्रत्येक प्रोडक्ट के लिए अकाउंट्स की संख्या $sum: 1 से गिनता है (हर डॉक्यूमेंट के लिए 1 जोड़ना)

3. $sort स्टेज औसत लिमिट के आधार पर परिणामों को अवरोही क्रम में सजाता है

आउटपुट:

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संचालित द्वारा

परिणाम दर्शाते हैं कि Commodity प्रोडक्ट्स सबसे अधिक औसत अकाउंट लिमिट से जुड़े हैं, जबकि CurrencyService सबसे कम से।

इस प्रकार का विश्लेषण प्रोडक्ट ऑफ़रिंग्स और ग्राहक खर्च क्षमता के बीच सहसंबंधों की पहचान में सहायक होता है।

अपनी एप्लिकेशंस में, आप इसी तकनीक का उपयोग श्रेणी के अनुसार बिक्री, फीचर के अनुसार यूज़र एंगेजमेंट, या मॉड्यूल के अनुसार एरर्स का विश्लेषण करने में कर सकते हैं—जहाँ भी आपको व्यक्तिगत रिकॉर्ड्स की बजाय समूहों में डेटा का सार चाहिए।

कलेक्शंस को $lookup और $unwind के साथ जोड़ना

जब आपका डेटा कई कलेक्शंस में फैला हो, तो $lookup उसे साथ लाने में मदद करता है। आइए ग्राहकों को उनके संबंधित अकाउंट्स के साथ ढूँढें:

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'])}")संचालित द्वारा

यह पाइपलाइन कलेक्शंस के पार संबंधित डेटा को जोड़ने का तरीका दिखाती है:

$match स्टेज यूज़रनेम से एक विशिष्ट ग्राहक खोजता है
$lookup स्टेज accounts कलेक्शन के साथ लेफ्ट आउटर जॉइन करता है:

from: किस कलेक्शन से जॉइन करना है यह बताता है
localField: वर्तमान कलेक्शन (customers) का मेल कराने वाला फ़ील्ड
foreignField: टार्गेट कलेक्शन (accounts) का मेल कराने वाला फ़ील्ड
as: नए एरे फ़ील्ड का नाम जो मिलते-जुलते डॉक्यूमेंट्स रखेगा

3. $project स्टेज आउटपुट को केवल प्रासंगिक फ़ील्ड्स दिखाने के लिए शेप करता है

आउटपुट:

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संचालित द्वारा

आउटपुट Elizabeth Ray के वित्तीय पोर्टफोलियो का समग्र दृश्य देता है—एक ही क्वेरी परिणाम में उसके सभी छह अकाउंट्स और उनसे जुड़े प्रोडक्ट्स।

यह बिना कई क्वेरियों या क्लाइंट-साइड जॉइन्स के आपके डेटा संबंधों का 360-डिग्री दृश्य सक्षम करता है।

अपनी एप्लिकेशंस में देखें कि आपने नॉर्मलाइज़ेशन के लिए डेटा कहाँ-कहाँ कलेक्शंस में बाँटा है लेकिन विश्लेषण या डिस्प्ले हेतु उसे फिर से एक साथ लाना होता है। सामान्य उदाहरणों में गतिविधि इतिहास के साथ यूज़र प्रोफ़ाइल, इन्वेंटरी स्थिति के साथ उत्पाद, या संबंधित टिप्पणियों के साथ कंटेंट शामिल हैं।

जटिल विश्लेषण के लिए कई स्टेज को संयोजित करना

अब एक अधिक जटिल व्यावसायिक प्रश्न लें: “सेवा टियर के अनुसार समूहित करने पर ग्राहकों के लिए औसत अकाउंट लिमिट्स क्या हैं?” इसके लिए हमें ग्राहक डॉक्यूमेंट्स के भीतर एक नेस्टेड ऑब्जेक्ट (tier_and_details) को सँभालना होगा और फिर इस जानकारी को accounts कलेक्शन से जोड़ना होगा।

पहले, customers कलेक्शन में जिस संरचना से हम डील कर रहे हैं, उसे समझने के लिए एक सैंपल ग्राहक का tier_and_details फ़ील्ड और accounts फ़ील्ड देखें। tier_and_details एक ऑब्जेक्ट है जहाँ प्रत्येक कुंजी किसी टियर सब्सक्रिप्शन का आइडेंटिफायर है, और मान में टियर नाम जैसी डिटेल्स होती हैं। accounts फ़ील्ड उस ग्राहक से जुड़े अकाउंट IDs का एक एरे है।

# 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']}")संचालित द्वारा

आउटपुट:

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]संचालित द्वारा

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']}")संचालित द्वारा

यह पाइपलाइन जटिल विश्लेषण को आठ प्रबंधनीय चरणों में बाँटती है:

$project: हम tier_and_details ऑब्जेक्ट को रूपांतरित करने से शुरू करते हैं। $objectToArray ऑपरेटर इस ऑब्जेक्ट को key-value पेयर्स की एरे (tiers_array) में बदल देता है। यह आवश्यक है क्योंकि $unwind जैसे स्टेज एरे पर काम करते हैं। महत्वपूर्ण रूप से, हम ग्राहक डॉक्यूमेंट से accounts एरे को customer_account_ids के रूप में और ग्राहक के _id को भी स्पष्ट रूप से आगे ले जाते हैं।
$unwind: यह स्टेज tiers_array को डीकंस्ट्रक्ट करता है, जिससे किसी ग्राहक के हर टियर एंट्री के लिए अलग डॉक्यूमेंट बनता है। प्रत्येक नए डॉक्यूमेंट में customer_account_ids और मूल ग्राहक _id बना रहता है।
$project: हम डॉक्यूमेंट को इस तरह रिशेप करते हैं कि टियर नाम (जैसे "Bronze") नेस्टेड संरचना ($tiers_array.v.tier) से स्पष्ट रूप से निकले, और स्पष्टता हेतु ग्राहक _id का नाम customer_id रखा जाए। customer_account_ids को भी पास-थ्रू करते हैं।
$lookup: यहीं हम accounts कलेक्शन से जॉइन करते हैं। हम ग्राहक डॉक्यूमेंट के अकाउंट नंबरों की एरे customer_account_ids को localField के रूप में उपयोग करते हैं। foreignField accounts कलेक्शन का account_id है। MongoDB वे सभी अकाउंट्स ढूँढेगा जिनका account_id customer_account_ids एरे में मौजूद है, और उन्हें matched_account_details फ़ील्ड में एरे के रूप में जोड़ देगा।
$unwind: हम matched_account_details को अनवाइंड करते हैं। अब यदि किसी ग्राहक-टियर संयोजन से कई अकाउंट जुड़े थे, तो हमें प्रत्येक विशिष्ट अकाउंट के लिए अलग डॉक्यूमेंट मिलता है जो उस ग्राहक और टियर से संबद्ध है।
$group: हम डॉक्यूमेंट्स को टियर के आधार पर समूहित करते हैं। प्रत्येक टियर के लिए, हम जुड़े अकाउंट डिटेल्स के limit पर $avg का उपयोग कर avg_limit निकालते हैं। $sum: 1 से total_accounts_in_tier गिनते हैं। साथ ही customer_id के साथ $addToSet का उपयोग कर प्रत्येक टियर में अनोखे ग्राहकों के IDs एकत्र करते हैं।
$project: आउटपुट का अंतिम शेपिंग। हम _id (जो ग्रुप स्टेज से टियर नाम है) का नाम tier_name रखते हैं। unique_customers_in_tier की गिनती पाने हेतु $size का उपयोग करते हैं।
$sort: हम परिणामों को average_account_limit के अवरोही क्रम में क्रमबद्ध करते हैं, ताकि सबसे अधिक औसत लिमिट वाला टियर पहले दिखे।

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संचालित द्वारा

अब परिणाम थोड़ा अलग अनुक्रम दिखाते हैं: औसतन Silver टियर ग्राहकों के अकाउंट्स की लिमिट सबसे अधिक है, उसके बाद Bronze, फिर Platinum, और अंत में Gold।

इस प्रकार की अंतर्दृष्टि ग्राहक सेगमेंट्स को समझने में मूल्यवान होती है। उदाहरण के लिए, कोई वित्तीय संस्था यह जाँच सकती है कि Silver और Bronze टियर्स की औसत लिमिट इतनी अधिक क्यों है, या मार्केटिंग अभियानों/प्रीमियम सेवाओं को उसी अनुसार अनुकूलित कर सकती है।

नेस्टेड टियर डेटा को अनलॉक करने में $objectToArray का उपयोग प्रमुख था, और customer_account_ids को सावधानी से पास करना सुनिश्चित करता है कि हमारा $lookup ग्राहकों को उनके विशिष्ट अकाउंट्स से जोड़ सके।

जब आपको अपने डेटासेट्स में नेस्टेड ऑब्जेक्ट्स को aggregation में उपयोग करना हो (जैसे user preferences, product attributes, या configuration settings), तो $objectToArray तकनीक याद रखें।

हमेशा सुनिश्चित करें कि बाद के स्टेजेज़—विशेषकर $lookup ऑपरेशंस—के लिए आवश्यक किसी भी फ़ील्ड को आप अपने $project स्टेजेज़ में स्पष्ट रूप से शामिल करें। जटिल डेटा को चरणबद्ध तरीके से तोड़ने का यह संरचित दृष्टिकोण MongoDB के भीतर ही अर्थपूर्ण इनसाइट्स निकालने में मदद करता है।

एडवांस्ड पाइपलाइन पैटर्न

क्रमिक ट्रांसफ़ॉर्मेशंस से परे, MongoDB का aggregation फ़्रेमवर्क जटिल विश्लेषणात्मक क्वेरियों को संभालने के लिए परिष्कृत पैटर्न प्रदान करता है। जब आपको एक साथ कई कोणों से डेटा का विश्लेषण करना हो या दस्तावेज़ों के रोलिंग सेट के आधार पर गणनाएँ करनी हों, तो ये उन्नत पैटर्न सीधे डेटाबेस के भीतर शक्तिशाली समाधान देते हैं।

$facet स्टेज आपको एक ही स्टेज में, समान इनपुट डॉक्यूमेंट सेट का उपयोग करके, कई aggregation सब-पाइपलाइन्स चलाने देता है। कल्पना करें कि आपको किसी प्रोडक्ट कलेक्शन से प्रोडक्ट्स को प्राइस रेंज के अनुसार वर्गीकृत भी करना है और साथ ही सबसे लोकप्रिय शीर्ष 5 प्रोडक्ट ब्रांड्स की सूची भी बनानी है। $facet समानांतर में इन अलग-अलग विश्लेषणात्मक दृष्टिकोणों को संभालता है।

$facet के भीतर हर सब-पाइपलाइन इनपुट डॉक्यूमेंट्स पर स्वतंत्र रूप से कार्य करती है और अपना आउटपुट डॉक्यूमेंट्स का एरे बनाती है। इसका अर्थ है कि आप—एक व्यापक डैशबोर्ड के लिए आवश्यक—अलग-अलग मैट्रिक्स या सारांश एक ही डेटाबेस क्वेरी में इकट्ठे कर सकते हैं।

उदाहरण के लिए, आप एक ही यूज़र डेटासेट से एक साथ कुल सक्रिय उपयोगकर्ताओं की गिनती, सब्सक्रिप्शन टियर के अनुसार उपयोगकर्ताओं का ब्रेकडाउन, और हाल ही में जुड़े उपयोगकर्ताओं की सूची प्राप्त कर सकते हैं। इससे जटिल व्यूज़ के लिए डेटा रिट्रीवल को सुगम करते हुए, कई डेटाबेस कॉल्स के ओवरहेड के बिना समृद्ध, बहुआयामी रिपोर्ट्स बनाना संभव होता है।

विंडो फ़ंक्शंस के साथ काल-आधारित विश्लेषण

MongoDB 5.0 में पेश किए गए विंडो फ़ंक्शंस उन दस्तावेज़ों के सेट पर गणनाएँ करते हैं जो वर्तमान डॉक्यूमेंट से संबंधित होते हैं—इसे “विंडो” कहा जाता है।

यह टाइम-सीरीज़ डेटा या किसी भी क्रमबद्ध डेटासेट के लिए उपयोगी है जहाँ पड़ोसी डॉक्यूमेंट्स का संदर्भ महत्वपूर्ण होता है। उदाहरण के लिए, आप बिक्री का 7-दिवसीय मूविंग एवरेज निकालना चाहते हैं या प्रत्येक ग्राहक के लिए समय के साथ लेनदेन का संचयी योग ढूँढना चाहते हैं।

विंडो फ़ंक्शंस आम तौर पर $setWindowFields स्टेज के भीतर उपयोग किए जाते हैं। यह स्टेज आपको पार्टिशन (दस्तावेज़ समूह, जैसे प्रति प्रोडक्ट बिक्री) और उनके भीतर सॉर्ट क्रम (जैसे तिथि के अनुसार) परिभाषित करने देता है।

फिर आप परिभाषित विंडो (उदा., पिछले 3 डॉक्यूमेंट्स और वर्तमान डॉक्यूमेंट) पर $avg, $sum, $min, $max, या $derivative या $integral जैसे विशेष फ़ंक्शंस लगा सकते हैं।

मान लीजिए उत्पाद बिक्री का रनिंग टोटल निकालना है। $setWindowFields का उपयोग करने वाली पाइपलाइन कुछ इस प्रकार हो सकती है:

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))संचालित द्वारा

इस वैचारिक पाइपलाइन में, प्रत्येक इलेक्ट्रॉनिक बिक्री के लिए एक running_total_sales फ़ील्ड जोड़ा जाता है। यह फ़ील्ड उस उत्पाद के लिए सबसे शुरुआती बिक्री से लेकर वर्तमान बिक्री तक sale_amount के योग का प्रतिनिधित्व करता है।

ऐसी गणनाएँ जो पहले जटिल क्लाइंट-साइड लॉजिक या कई क्वेरियों की माँग करती थीं, अब सीधे डेटाबेस में की जा सकती हैं—ट्रेंड एनालिसिस या अवधि-दर-अवधि तुलना के लिए एप्लिकेशन कोड को सरल बनाते हुए और प्रदर्शन सुधारते हुए।

निष्कर्ष

MongoDB aggregation pipelines आपके डेटाबेस के भीतर ही डेटा प्रोसेस करने की एक संरचित विधि प्रदान करती हैं। अलग-अलग ऑपरेशनल स्टेज को जोड़कर आप रिकॉर्ड्स फ़िल्टर करने, डॉक्यूमेंट्स को रिशेप करने, जानकारी को समूहित करने, और विभिन्न कलेक्शंस से डेटा संयोजित करने जैसे जटिल रूपांतरण कर सकते हैं।

यह दृष्टिकोण आपके डेटा वर्कफ़्लोज़ को परिष्कृत करने में मदद करता है, जैसा कि PyMongo उदाहरणों में दिखाया गया है—बड़े डेटासेट्स से अर्थ निकालना तेज़ बनाता है। स्रोत के पास डेटा प्रोसेस करना अक्सर विश्लेषणात्मक क्वेरियों के लिए बेहतर प्रदर्शन देता है।

Aggregation pipelines में दक्षता आपको अनेक डेटा विश्लेषण चुनौतियों के लिए तैयार करती है। जो आपने सीखा है उसे लागू करने के लिए, आप Building a Retail Data Pipeline जैसा हैंड्स-ऑन प्रोजेक्ट आज़मा सकते हैं।

डेटा सिस्टम्स पर व्यापक दृष्टिकोण और इस पर कि बड़े परिदृश्य में MongoDB कौशल कैसे फिट होते हैं, के लिए Data Engineer with Python ट्रैक गहन शिक्षण प्रदान करता है। यदि आप करियर प्रगति की ओर देख रहे हैं, तो सामान्य MongoDB Interview Questions की समीक्षा करना भी सामान्य समस्या-समाधान अपेक्षाओं को समझने में सहायक हो सकता है।

MongoDB aggregation pipeline FAQs

MongoDB aggregation pipelines क्या हैं?

MongoDB aggregation pipelines डेटा विश्लेषण के लिए एक फ़्रेमवर्क हैं। ये चरणों की श्रृंखला के माध्यम से डॉक्यूमेंट्स को प्रोसेस करती हैं और उन्हें ट्रांसफॉर्म कर गणना किए गए परिणाम लौटाती हैं।

Aggregation pipelines डेटा प्रोसेसिंग में कैसे सुधार लाती हैं?

ये MongoDB के भीतर सीधे जटिल डेटा ट्रांसफ़ॉर्मेशन और विश्लेषण की अनुमति देती हैं, डेटा ट्रांसफ़र को कम करती हैं और बेहतर प्रदर्शन के लिए डेटा को उसके स्रोत के पास प्रोसेस करती हैं।

MongoDB aggregation pipeline में सामान्य स्टेज कौन-से हैं?

सामान्य स्टेज में फ़िल्टरिंग के लिए $match, डॉक्यूमेंट्स के रिशेपिंग के लिए $project, डेटा एग्रीगेशन के लिए $group, और कलेक्शंस को जोड़ने के लिए $lookup शामिल हैं।

क्या मैं Python के साथ MongoDB aggregation pipelines का उपयोग कर सकता/सकती हूँ?

हाँ, यह ट्यूटोरियल PyMongo—MongoDB का आधिकारिक Python ड्राइवर—का उपयोग करके MongoDB aggregation pipelines बनाना और चलाना दिखाता है।

Aggregation pipelines में स्टेज क्रम महत्वपूर्ण क्यों है?

प्रदर्शन के लिए स्टेज का क्रम महत्वपूर्ण है। शुरुआती चरण में $match से डेटा फ़िल्टर करना, बाद के अधिक जटिल स्टेजेज़ द्वारा प्रोसेस किए जाने वाले डॉक्यूमेंट्स की मात्रा घटा देता है।

लेखक

Bexruz (Bex) Tuychiev

मैं 2 से अधिक वर्षों के अनुभव वाला डेटा साइंस कंटेंट क्रिएटर हूँ और Medium पर सबसे बड़े फॉलोइंग्स में से एक रखता हूँ। मुझे एआई और एमएल पर विस्तार से लेख लिखना पसंद है, जिसमें हल्का-सा व्यंग्यात्मक अंदाज रहता है—क्योंकि कुछ तो करना पड़ता है ताकि वे कम उबाऊ लगें। मैंने 130 से अधिक लेख और एक DataCamp कोर्स तैयार किया है, और एक और बन रहा है। मेरा कंटेंट 50 लाख से अधिक लोगों ने देखा है, जिनमें से 20 हज़ार लोग Medium और LinkedIn पर फॉलोअर्स बन गए।

विषय

MongoDB डेटा इंजीनियरिंग

प्रौद्योगिकियाँ

विषय

Python में PyMongo के साथ MongoDB Aggregation Pipeline ट्यूटोरियल

AI के साथ खोजें

MongoDB में Aggregation Pipelines क्या हैं?

पाइपलाइन आर्किटेक्चर और डॉक्यूमेंट फ्लो

स्टेज की प्रकारिकी और कार्यात्मक श्रेणियाँ

ट्यूटोरियल के लिए वातावरण तैयार करना

PyMongo इंस्टॉल करना

सैंपल डेटासेट तक पहुँच

MongoDB से कनेक्ट करना

MongoDB Aggregation Pipelines का गहन विवरण

$match के साथ डेटा फ़िल्टर करना

$project, $sort, $limit, और $skip से डॉक्यूमेंट्स का रिशेपिंग

$group के साथ ग्रुपिंग और एग्रीगेशन

कलेक्शंस को $lookup और $unwind के साथ जोड़ना

जटिल विश्लेषण के लिए कई स्टेज को संयोजित करना

एडवांस्ड पाइपलाइन पैटर्न

$facet के साथ बहुआयामी एनालिटिक्स

विंडो फ़ंक्शंस के साथ काल-आधारित विश्लेषण

निष्कर्ष

MongoDB aggregation pipeline FAQs

विषय

शीर्ष DataCamp कोर्स

Python में MongoDB परिचय

डेटा इंजीनियर में Python

NoSQL कॉन्सेप्ट्स

प्रौद्योगिकियाँ

विषय

शीर्ष DataCamp कोर्स

Python में MongoDB परिचय

डेटा इंजीनियर में Python

NoSQL कॉन्सेप्ट्स