MongoDB-aggregatiepijplijn-tutorial in Python met PyMongo

Werken met grote datasets brengt vaak de uitdaging met zich mee om zinvolle patronen en inzichten te halen, terwijl je de performance bewaakt. Als je applicaties data in MongoDB opslaan, kunnen complexe queries en transformaties direct in de database uitvoeren veel sneller zijn dan data naar externe analyse-tools verplaatsen. Aggregatiepijplijnen in MongoDB bieden een oplossing door je data te laten verwerken, transformeren en analyseren waar die zich bevindt.

Je kunt aangepaste dataverwerkingsworkflows maken door eenvoudige bewerkingen achter elkaar te koppelen. Elke stage in de pijplijn transformeert documenten en geeft de resultaten door aan de volgende stage. Je moet bijvoorbeeld misschien records filteren op datumbereik, ze groeperen op categorie, statistische maten berekenen en de output formatteren—allemaal uitgevoerd via één databaseoperatie die data dicht bij de bron verwerkt.

In dit artikel leer je aggregatiepijplijnen bouwen om veelvoorkomende data-uitdagingen op te lossen, met alle voorbeelden gedemonstreerd in PyMongo (MongoDB’s officiële Python-client). 

Hoewel dit artikel focust op aggregatiepijplijnen, als je nieuw bent met MongoDB in Python, is de cursus Introduction to MongoDB in Python een uitgebreid startpunt. Met dit artikel leg je genoeg basis om deze aggregatieconcepten zelf te vertalen naar MongoDB’s querytaal, al dan niet met hulp van taalmodellen.

Wat zijn aggregatiepijplijnen in MongoDB?

Stel dat je klantreviews over meerdere producten wilt analyseren om tevredenheidstrends te begrijpen. Traditionele queries kunnen de data ophalen, maar ze helpen niet met het combineren, analyseren en transformeren van die informatie tot bruikbare samenvattingen. 

De aggregatiepijplijnen van MongoDB lossen dit op door een gestructureerde manier te bieden om data te verwerken via een reeks bewerkingen die op elkaar voortbouwen.

Pijplijnarchitectuur en documentstroom

Zie aggregatiepijplijnen als lopende banden voor je data. Elk document uit je collectie gaat aan de ene kant de pijplijn in en passeert verschillende stations (stages) waar het wordt gefilterd, getransformeerd, gegroepeerd of verrijkt. 

De output van de ene stage wordt de input voor de volgende, waardoor je complexe datatransformaties kunt opdelen in kleinere, beheersbare stappen.

Deze pijplijnen gebruiken een declaratieve aanpak; je specificeert wat je in elke stage wilt, niet hoe je het moet berekenen. Deze aanpak maakt je verwerkingsintentie duidelijk en laat MongoDB de uitvoering afhandelen. De database kan vervolgens optimalisaties toepassen op basis van je pijplijnstructuur.

De volgorde van stages is belangrijk in je pijplijnontwerp. Documenten vroeg filteren (voor groeperen of complexe berekeningen) vermindert de hoeveelheid data die door de pijplijn stroomt. 

Deze aanpak kan de performance aanzienlijk verbeteren bij grote collecties. Een goed gestructureerde pijplijn verwerkt alleen de data die je nodig hebt voor je eindresultaten.

• Documentstroom: Documenten komen binnen → transformeren door stages → resultaten komen eruit
• Declaratief ontwerp: Geef aan wat je wilt, niet hoe je het berekent
• Geordende verwerking: De output van elke stage gaat direct door naar de volgende
• Performancevoordeel: Vroeg filteren verkleint het datavolume in latere stages

Stage-typologie en functionele categorieën

Aggregatiestages in MongoDB vallen in vier hoofdcategorieën op basis van hun doel. Filterstages zoals $match werken als queries en selecteren alleen documenten die aan specifieke criteria voldoen. Dit helpt je dataset te verkleinen voordat je complexere bewerkingen uitvoert.

Herschikkingsstages transformeren de documentstructuur. Met $project of $addFields kun je velden opnemen, uitsluiten of hernoemen, of berekende velden maken op basis van bestaande waarden. Deze stages helpen documenten te vereenvoudigen door alleen relevante informatie te behouden en berekende waarden toe te voegen die nodig zijn voor analyse.

Als je meerdere documenten moet combineren op basis van gedeelde kenmerken, komen groeperingsstages in beeld. De $group-stage is hier het werkpaard, waarmee je aantallen, sommen, gemiddelden en andere geaggregeerde waarden over groepen documenten kunt berekenen. Dit verandert duizenden individuele records in zinvolle samenvattingen die jouw analytische vragen beantwoorden.

Om je dataplaatje compleet te maken, kun je met join-stages zoals $lookup informatie uit meerdere collecties combineren. Zo kun je documenten verrijken met gerelateerde data, vergelijkbaar met SQL-joins maar aangepast aan het documentmodel van MongoDB. De mogelijkheid om data over collecties heen te refereren helpt bij het behouden van normalisatie, terwijl je toch complete resultaten in één operatie krijgt.

• Filteren: Selecteer specifieke documenten met $match op basis van criteria
• Herschikken: Transformeer de documentstructuur met $project en $addFields
• Groeperen: Combineer en vat samen met $group voor geaggregeerde waarden
• Joinen: Verbind gerelateerde data over collecties met $lookup

Voor een complete referentie van mogelijkheden en operators kun je de MongoDB Aggregation Pipeline Manual raadplegen.

De omgeving instellen voor de tutorial

Voordat je in aggregatiepijplijnen in MongoDB duikt, heb je een werkende omgeving nodig met PyMongo en toegang tot een voorbeeld-dataset. In deze sectie doorlopen we het instelproces met de dataset sample_analytics, die financiële data bevat en perfect is om aggregatieconcepten te demonstreren.

PyMongo installeren

PyMongo is de officiële Python-driver van MongoDB. Je kunt deze installeren met pip op zowel macOS als Windows:

# Install PyMongo using pip
pip install pymongo

# Or if you're using conda
conda install -c conda-forge pymongoMogelijk gemaakt door

• Versiecompatibiliteit: PyMongo 4.x werkt met Python 3.7+
• Afhankelijkheden: De installer handelt vereiste dependencies automatisch af
• Platformondersteuning: Werkt identiek op Windows, macOS en Linux

Toegang tot de voorbeeld-dataset

MongoDB biedt voorbeeld-datasets die je kunt gebruiken zonder zelf data te maken. De eenvoudigste aanpak is MongoDB Atlas (de cloudversie) gebruiken:

1. Maak een gratis MongoDB Atlas-account aan op mongodb.com/cloud/atlas
2. Stel een free tier-cluster in

2.1. Kies de gratis versie (free forever)

2.2. Geef je cluster een naam

2.3. Klik op “Deploy”

2.4. Kopieer je gebruikersnaam en wachtwoord naar een veilige plek

2.5. Klik op “Create database user”

2.6. Kies “Drivers” als je verbindingsmethode

2.7. Kopieer de connection URI voor de volgende stap

3.1. Klik op de drie puntjes in je clusteroverzicht

3.2. Kies “Load sample dataset”

3.3. Selecteer in de lijst sample_analytics en wacht tot deze geladen is.

Voor lokale MongoDB-installaties kun je de voorbeeld-datasets laden met:

# Download and restore the sample dataset
python -m pip install pymongo[srv]
python -c "from pymongo import MongoClient; MongoClient().admin.command('getParameter', '*')"Mogelijk gemaakt door

• Datagrootte: De sample_analytics-dataset is klein genoeg (~10MB) voor de free tier
• Collectiestructuur: Bevat klanten- en accountscollecties met relatiedata
• Daterealiteit: Gebaseerd op realistische financiële patronen voor betekenisvolle aggregatievoorbeelden

Verbinden met MongoDB

Nu MongoDB en de voorbeeld-dataset klaarstaan, schrijven we de code om te verbinden en onze setup te verifiëren:

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())Mogelijk gemaakt door

Output:

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']}Mogelijk gemaakt door

• Verificatie: De code bevestigt je verbinding en datasettoegang
• Documentvoorbeeld: Toont de structuur van documenten waarmee je gaat werken

Wanneer je deze code draait, zou je output moeten zien met de documentaantallen en een preview van zowel klant- als accountdocumenten. Dit bevestigt dat je omgeving klaar is voor de aggregatiepijplijnvoorbeelden die we in de volgende secties verkennen.

Voor een introductie in PyMongo en lokale MongoDB-setup, bekijk onze beginnerstutorial PyMongo.

MongoDB-aggregatiepijplijnen uitgediept

Nu we onze verbinding hebben bevestigd en de documentstructuur hebben bekeken, duiken we in het aggregatiekader van MongoDB. Deze krachtige functionaliteit laat ons data direct in de database verwerken en transformeren. In deze sectie behandelen we de meest voorkomende pijplijnstages met praktische voorbeelden op basis van onze voorbeeld-dataset.

Data filteren met $match

De $match-stage filtert documenten op basis van opgegeven criteria. Zie het als een manier om je te richten op precies de data die je belangrijk vindt voordat je complexere bewerkingen doet.

Laten we alle premium-accounts vinden met een limiet boven $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])Mogelijk gemaakt door

Deze pijplijn gebruikt de $match-stage met een vergelijkingoperator $gt (groter dan) om accounts te filteren. Het werkt net als de find()-methode maar dan binnen de context van een pijplijn. De query bekijkt elk document in de accounts-collectie en behoudt alleen die waarvan het veld limit groter is dan 9000.

Output:

Found 1701 premium accounts
{'_id': ObjectId('5ca4bbc7a2dd94ee5816238d'), 'account_id': 557378, 'limit': 10000, 'products': ['InvestmentStock', 'Commodity', 'Brokerage', 'CurrencyService']}Mogelijk gemaakt door

Uit de resultaten blijkt dat de pijplijn 1701 premium-accounts identificeerde van in totaal 1746 accounts in onze database. Deze filterstap versmalt onze focus, waardoor vervolg­analyse efficiënter en gerichter wordt. In je eigen projecten kun je deze techniek gebruiken om je te richten op actieve gebruikers, transacties boven een bepaald bedrag of producten in een specifieke categorie, voordat je dieper in die documenten duikt.

Documenten herschikken met $project, $sort, $limit en $skip

Deze stages helpen je bepalen welke velden je opneemt en hoe je je resultaten ordent. Hier is een voorbeeld dat de top 5 accounts met de hoogste limieten ophaalt en alleen essentiële informatie toont:

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)Mogelijk gemaakt door

Deze pijplijn heeft drie stages die samenwerken:

1. De $project-stage herschikt elk document door:

• Het veld _id uit te sluiten (op 0 zetten)
• De velden account_id en limit op te nemen (op 1 zetten)
• Een nieuw veld product_count te maken dat de operator $size (één van vele operators voor aggregatiepijplijnen) gebruikt om items in de products-array te tellen

2. De $sort-stage ordent resultaten op het veld limit in aflopende volgorde (-1)

3. De $limit-stage houdt na het sorteren alleen de eerste 5 documenten over

Output:

{'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}Mogelijk gemaakt door

De output toont de vijf accounts met de hoogste limieten, allemaal op $10.000. We zien ook hoeveel producten elk account heeft via ons berekende veld product_count. 

Dit maakt een schonere datavoorstelling mogelijk die precies focust op wat nodig is, in plaats van alle velden terug te geven. Bij het bouwen van dashboards of rapportages kun je soortgelijke technieken gebruiken om alleen de meest relevante informatie te tonen, waardoor datatransfer afneemt en de UI eenvoudiger wordt.

Groeperen en aggregeren met $group

De $group-stage is waar aggregatie echt uitblinkt. Je kunt documenten categoriseren en metrics per groep berekenen. Laten we de gemiddelde accountlimiet per producttype vinden:

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']}")Mogelijk gemaakt door

Deze pijplijn gebruikt meerdere operators om productdata te analyseren:

1. De $unwind-operator splitst documenten met meerdere producten in afzonderlijke documenten. Een account met ["Derivatives", "InvestmentStock"] wordt bijvoorbeeld twee documenten, één per product.
2. De $group-stage doet vervolgens:

• Groeperen op productnaam (het _id-veld in $group bepaalt de groeperingssleutel)
• Het gemiddelde van de limiet voor elk product berekenen met de $avg-accumulator
• Het aantal accounts per product tellen met $sum: 1 (1 optellen per document)

3. De $sort-stage ordent resultaten op gemiddeld limiet in aflopende volgorde

Output:

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: 742Mogelijk gemaakt door

De resultaten laten zien dat Commodity-producten gekoppeld zijn aan de hoogste gemiddelde accountlimieten, terwijl CurrencyService-producten de laagste hebben. 

Dit soort analyses helpt correlaties te identificeren tussen productaanbod en bestedingscapaciteit van klanten. 

In je eigen applicaties kun je vergelijkbare technieken gebruiken om verkoop per categorie te analyseren, engagement per feature of fouten per module—elk scenario waarin je data over groepen wilt samenvatten in plaats van individuele records te bekijken.

Collecties joinen met $lookup en $unwind

Wanneer je data over meerdere collecties verspreid is, helpt $lookup om het samen te brengen. Laten we klanten vinden met hun bijbehorende accounts:

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'])}")Mogelijk gemaakt door

Deze pijplijn laat zien hoe je gerelateerde data over collecties kunt joinen:

1. De $match-stage vindt een specifieke klant op gebruikersnaam
2. De $lookup-stage voert een left outer join uit met de accounts-collectie:

• from: specificeert met welke collectie je joinet
• localField: het veld in de huidige collectie (customers) om op te matchen
• foreignField: het veld in de doelcollectie (accounts) om mee te matchen
• as: de naam voor het nieuwe arrayveld dat de overeenkomende documenten bevat

3. De $project-stage vormt de output zodat alleen relevante velden zichtbaar zijn

Output:

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, InvestmentStockMogelijk gemaakt door

De output geeft een compleet beeld van de financiële portefeuille van Elizabeth Ray, met al haar zes accounts en de bijbehorende producten in één queryresultaat. 

Dit maakt een volledige 360-gradenweergave van je datarelaties mogelijk zonder meerdere queries of client-side joins. 

Bedenk voor je eigen applicaties waar je data over collecties hebt opgesplitst om te normaliseren, maar het weer moet samenbrengen voor analyse of weergave. Veelvoorkomende voorbeelden zijn gebruikersprofielen met activiteitsgeschiedenis, producten met voorraadstatus of content met gerelateerde reacties.

Meerdere stages combineren voor complexe analyses

Laten we nu een meer betrokken businessvraag aanpakken: “Wat zijn de gemiddelde accountlimieten voor klanten, gegroepeerd op hun servicetier?” Hiervoor moeten we een genest object in onze klantdocumenten (tier_and_details) verwerken en deze informatie vervolgens koppelen aan de collectie accounts.

Om eerst de structuur te begrijpen waarmee we in de collectie customers werken, bekijken we het veld tier_and_details en het veld accounts voor een voorbeeldklant. Het veld tier_and_details is een object waarbij elke sleutel een id is voor een tierabonnement en de waarde details zoals de tiernaam bevat. Het veld accounts is een array met account-id’s die aan die klant zijn gekoppeld.

# 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']}")Mogelijk gemaakt door

Output:

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]Mogelijk gemaakt door

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']}")Mogelijk gemaakt door

Deze pijplijn breekt de complexe analyse op in acht beheersbare stappen:

1. $project: We beginnen met het transformeren van het object tier_and_details. De $objectToArray-operator zet dit object om in een array van key-valueparen (tiers_array). Dit is essentieel omdat stages zoals $unwind op arrays werken. Cruciaal is dat we ook expliciet de accounts-array uit het klantdocument meenemen als customer_account_ids en de _id van de klant.
2. $unwind: Deze stage deconstrueert de tiers_array en maakt een apart document voor elke tiervermelding die een klant kan hebben. Elk nieuw document bevat nog steeds de customer_account_ids en de oorspronkelijke klant-_id.
3. $project: We herschikken het document om de tiernaam (bijv. "Bronze") duidelijk te extraheren uit de geneste structuur ($tiers_array.v.tier) en hernoemen de klant-_id naar customer_id voor duidelijkheid. De customer_account_ids worden doorgegeven.
4. $lookup: Hier joinen we met de collectie accounts. We gebruiken de customer_account_ids (de array met accountnummers uit het klantdocument) als localField. Het foreignField is account_id uit de collectie accounts. MongoDB vindt alle accounts waarvan  account_id is present in the customer_account_ids array, adding them as an array to the matched_account_details field.
5. $unwind: We ontvouwen matched_account_details. Als een klant-tiercombinatie aan meerdere accounts is gekoppeld, krijgen we nu een apart document voor elk specifiek account, gekoppeld aan die klant en tier.
6. $group: We groeperen de documenten op tier. Voor elke tier berekenen we de avg_limit met $avg over de limit uit de gekoppelde accountdetails. We tellen de total_accounts_in_tier met $sum: 1. We gebruiken ook $addToSet met customer_id om de unieke klant-id’s per tier te verzamelen.
7. $project: De uiteindelijke vormgeving van onze output. We hernoemen _id (de tiernaam uit de group-stage) naar tier_name. We gebruiken $size om het aantal unique_customers_in_tier te krijgen.
8. $sort: We ordenen de resultaten op average_account_limit` in aflopende volgorde, zodat de tier met de hoogste gemiddelde limiet bovenaan staat.

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: 99Mogelijk gemaakt door

De resultaten tonen nu een iets andere hiërarchie: klanten in de Silver-tier hebben gemiddeld accounts met de hoogste limieten, gevolgd door  Bronze, dan Platinum en tot slot Gold.

Dit soort inzicht is waardevol om klantsegmenten te begrijpen. Een financiële instelling kan deze informatie bijvoorbeeld gebruiken om te onderzoeken waarom Silver- en Bronze-tiers zulke hoge gemiddelde limieten hebben, of om marketingcampagnes te richten of premiumdiensten aan te bieden. 

Het gebruik van $objectToArray was de sleutel om de geneste tierdata te ontsluiten, en door zorgvuldig de customer_account_ids door te geven, kon onze $lookup klanten koppelen aan hun specifieke accounts.

Als je in je eigen datasets geneste objecten tegenkomt die je in aggregaties wilt gebruiken (zoals user preferences, product attributes of configuration settings), onthoud dan de $objectToArray-techniek. 

Zorg er altijd voor dat velden die nodig zijn voor latere stages, vooral voor $lookup-operaties, expliciet zijn opgenomen in je $project-stages. Deze gestructureerde aanpak om complexe data op te breken helpt om direct binnen MongoDB betekenisvolle inzichten te verkrijgen.

Geavanceerde pijplijnpatronen

Naast sequentiële transformaties biedt het aggregatiekader van MongoDB geavanceerde patronen om complexe analytische queries aan te pakken. Wanneer je data vanuit meerdere invalshoeken tegelijk moet analyseren of berekeningen wilt doen op basis van een schuivende set documenten, leveren deze geavanceerde patronen krachtige oplossingen direct in de database.

Multidimensionale analytics met $facet

De $facet-stage laat je meerdere aggregatie-subpijplijnen binnen één stage uitvoeren met dezelfde set inputdocumenten. Stel dat je producten wilt categoriseren op prijsklasse en tegelijkertijd de top 5 populairste productmerken uit dezelfde productcollectie wilt tonen. $facet behandelt deze verschillende analytische perspectieven parallel.

Elke subpijplijn binnen $facet werkt onafhankelijk op de inputdocumenten en produceert een eigen array met outputdocumenten. Dit betekent dat je in één databasequery verschillende metrics of samenvattingen kunt verzamelen—zoals nodig voor een allesomvattend dashboard—zonder extra overhead. 

Je kunt bijvoorbeeld tegelijkertijd een telling van het totaal aantal actieve gebruikers, een uitsplitsing van gebruikers per abonnementstier en een lijst met recent toegetreden gebruikers ophalen uit dezelfde gebruikersdataset. Dit maakt rijke, veelzijdige rapporten mogelijk zonder meerdere databasecalls, wat de dataverwerking voor complexe overzichten stroomlijnt.

Tijdsanalyse met windowfuncties

Geïntroduceerd in MongoDB 5.0: windowfuncties voeren berekeningen uit over een set documenten die gerelateerd zijn aan het huidige document, ook wel een “window” genoemd. 

Dit is nuttig voor tijdreeksdata of elke geordende dataset waarbij context van naburige documenten belangrijk is. Je wilt bijvoorbeeld een 7-daags voortschrijdend gemiddelde van verkopen berekenen of de cumulatieve som van transacties per klant in de tijd bepalen.

Windowfuncties worden doorgaans gebruikt binnen de $setWindowFields-stage. In deze stage definieer je partities (groepen documenten, zoals verkopen per product) en de sorteervolgorde binnen die partities (bijvoorbeeld op datum). 

Daarna kun je windowfuncties toepassen zoals $avg, $sum, $min, $max of gespecialiseerde functies zoals $derivative of $integral over een gedefinieerd window (bijv. de voorafgaande 3 documenten en het huidige document).

Stel dat je een running total van productverkopen wilt berekenen. Een pijplijn met $setWindowFields kan er zo uitzien:

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))Mogelijk gemaakt door

In deze conceptuele pijplijn wordt voor elke elektronische verkoop een veld running_total_sales toegevoegd. Dit veld representeert de som van sale_amount voor dat product vanaf de eerste verkoop tot en met de huidige verkoop. 

Dergelijke berekeningen, die voorheen complexe clientside-logica of meerdere queries vereisten, kunnen nu direct in de database worden uitgevoerd, wat applicatiecode vereenvoudigt en de performance verbetert voor trendanalyses of periode-over-periodevergelijkingen. 

Conclusie

Aggregatiepijplijnen in MongoDB bieden een gestructureerde methode om data direct in je database te verwerken. Door verschillende operationele stages te koppelen, kun je uitgebreide datamanipulaties uitvoeren, zoals records filteren, documenten herschikken, informatie groeperen en data uit verschillende collecties combineren. 

Deze aanpak helpt je dataworkflows te verfijnen, waardoor je sneller betekenis uit grote datasets haalt, zoals getoond in de PyMongo-voorbeelden. Data dicht bij de bron verwerken kan vaak leiden tot betere performance voor analytische queries.

Vaardig worden met aggregatiepijplijnen bereidt je voor op veel data-analyse-uitdagingen. Om te oefenen met wat je hebt geleerd, kun je een hands-on project proberen zoals Building a Retail Data Pipeline. 

Voor een breder perspectief op datasystemen en hoe MongoDB-vaardigheden in het grotere geheel passen, biedt de Data Engineer with Python-track diepgaande leerstof. Als je carrièrestappen wilt zetten, kan het doornemen van veelvoorkomende MongoDB-interviewvragen ook nuttig zijn om typische probleemoplossingsverwachtingen te begrijpen.