Tutorial sulla pipeline di aggregazione MongoDB in Python con PyMongo

Lavorare con dataset di grandi dimensioni spesso comporta la sfida di estrarre pattern e insight significativi mantenendo alte le prestazioni. Quando le tue applicazioni archiviano dati in MongoDB, eseguire query e trasformazioni complesse direttamente nel database può essere molto più veloce che spostare i dati in strumenti esterni di analisi. Le pipeline di aggregazione di MongoDB offrono una soluzione permettendoti di elaborare, trasformare e analizzare i dati lì dove risiedono.

Puoi creare flussi di elaborazione dati personalizzati collegando in sequenza operazioni semplici. Ogni fase nella pipeline trasforma i documenti e passa i risultati alla fase successiva. Per esempio, potresti dover filtrare i record per intervallo di date, raggrupparli per categoria, calcolare misure statistiche e formattare l’output — il tutto tramite un’unica operazione di database che elabora i dati vicino alla loro origine.

In questo articolo imparerai a costruire pipeline di aggregazione per risolvere sfide comuni sui dati, con tutti gli esempi dimostrati in PyMongo (il client Python ufficiale di MongoDB). 

Anche se questo articolo si concentra sulle pipeline di aggregazione, se sei alle prime armi con MongoDB in Python, il corso Introduction to MongoDB in Python è un ottimo punto di partenza. Con questo articolo acquisirai basi sufficienti per trasferire questi concetti di aggregazione nel linguaggio di query di MongoDB da solo o con l’aiuto di modelli linguistici.

Cosa sono le pipeline di aggregazione in MongoDB?

Immagina di dover analizzare le recensioni dei clienti su più prodotti per capire l’andamento della soddisfazione. Le query tradizionali potrebbero recuperare i dati, ma non aiutano a combinare, analizzare e trasformare quelle informazioni in riepiloghi utili. 

Le pipeline di aggregazione di MongoDB risolvono questo problema offrendo un modo strutturato per elaborare i dati tramite una serie di operazioni che si costruiscono una sull’altra.

Architettura della pipeline e flusso dei documenti

Pensa alle pipeline di aggregazione come a catene di montaggio per i tuoi dati. Ogni documento della tua collection entra da un’estremità della pipeline e attraversa varie stazioni (fasi) dove viene filtrato, trasformato, raggruppato o arricchito. 

L’output di una fase diventa l’input per la successiva, permettendoti di suddividere trasformazioni complesse in passaggi più piccoli e gestibili.

Queste pipeline usano un approccio dichiarativo: specifichi cosa vuoi ottenere a ogni fase invece di come calcolarlo. Questo rende chiare le tue intenzioni di elaborazione dei dati e lascia a MongoDB la gestione dei dettagli di esecuzione. Il database può quindi applicare diverse ottimizzazioni in base alla struttura della pipeline.

L’ordine delle fasi conta nella progettazione della pipeline. Filtrare i documenti all’inizio (prima di raggruppamenti o calcoli complessi) riduce la quantità di dati che fluiscono nella pipeline. 

Questo approccio può migliorare drasticamente le prestazioni quando si lavora con grandi collection. Una pipeline ben strutturata elabora solo i dati necessari per i risultati finali.

• Flusso dei documenti: I documenti entrano → si trasformano attraverso le fasi → emergono i risultati
• Progettazione dichiarativa: Specifica cosa vuoi, non come calcolarlo
• Elaborazione ordinata: L’output di ogni fase alimenta direttamente la fase successiva
• Beneficio prestazionale: Il filtraggio anticipato riduce il volume di dati nelle fasi successive

Tipologia delle fasi e categorie funzionali

Le fasi di aggregazione di MongoDB rientrano in quattro categorie principali in base al loro scopo. Le fasi di filtraggio come $match funzionano come query, selezionando solo i documenti che soddisfano criteri specifici. Questo aiuta a restringere il dataset prima di eseguire operazioni più complesse.

Le fasi di rimodellamento trasformano la struttura dei documenti. Usando $project o $addFields, puoi includere, escludere o rinominare campi, o creare campi calcolati basati su valori esistenti. Queste fasi aiutano a semplificare i documenti mantenendo solo le informazioni rilevanti e aggiungendo valori computati necessari per l’analisi.

Quando devi combinare più documenti in base a caratteristiche condivise, entrano in gioco le fasi di raggruppamento. La fase $group è il cavallo di battaglia qui, consentendoti di calcolare conteggi, somme, medie e altri valori aggregati su gruppi di documenti. Questo trasforma migliaia di record individuali in riepiloghi significativi che rispondono alle tue domande analitiche.

Per completare il quadro dei tuoi dati, le fasi di join come $lookup ti permettono di combinare informazioni da più collection. Questo ti consente di arricchire i documenti con dati correlati, in modo simile ai join SQL ma adattato al modello a documenti di MongoDB. La possibilità di fare riferimento a dati tra collection aiuta a mantenere una corretta normalizzazione pur fornendo risultati completi in un’unica operazione.

• Filtraggio: Seleziona documenti specifici con $match in base a criteri
• Rimodellamento: Trasforma la struttura dei documenti con $project e $addFields
• Raggruppamento: Combina e riassumi con $group per ottenere valori aggregati
• Join: Collega dati correlati tra collection con $lookup

Per una reference completa su funzionalità e operatori, puoi consultare il manuale della Aggregation Pipeline di MongoDB.

Preparare l’ambiente per il tutorial

Prima di addentrarti nelle pipeline di aggregazione MongoDB, ti serve un ambiente funzionante con PyMongo e l’accesso a un dataset di esempio. Questa sezione ti guida nel processo di setup con il dataset sample_analytics, che contiene dati finanziari perfetti per dimostrare i concetti di aggregazione.

Installare PyMongo

PyMongo è il driver Python ufficiale di MongoDB. Puoi installarlo con pip sia su macOS che su Windows:

# Install PyMongo using pip
pip install pymongo

# Or if you're using conda
conda install -c conda-forge pymongoOfferto da

• Compatibilità di versione: PyMongo 4.x funziona con Python 3.7+
• Gestione delle dipendenze: L’installer gestisce automaticamente le dipendenze richieste
• Supporto piattaforme: Funziona allo stesso modo su Windows, macOS e Linux

Accedere al dataset di esempio

MongoDB offre dataset di esempio che puoi usare senza creare i tuoi dati. L’approccio più semplice è usare MongoDB Atlas (la versione cloud):

1. Crea un account MongoDB Atlas gratuito su mongodb.com/cloud/atlas
2. Configura un cluster free tier

2.1. Scegli la versione gratuita per sempre

2.2. Dai un nome al tuo cluster

2.3. Clicca “Deploy”

2.4. Copia username e password in un posto sicuro

2.5. Clicca “Create database user”

2.6. Scegli “Drivers” come metodo di connessione

2.7. Copia la URI di connessione per il passo successivo

3.1. Clicca sui tre puntini nella vista dei tuoi cluster

3.2. Scegli “Load sample dataset”

3.3. Dall’elenco, seleziona sample_analytics e attendi il caricamento.

Per installazioni locali di MongoDB, puoi caricare i dataset di esempio usando:

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

• Dimensione dati: Il dataset sample_analytics è abbastanza piccolo (~10MB) da funzionare bene sul free tier
• Struttura delle collection: Contiene le collection customers e accounts con dati relazionali
• Realismo dei dati: Basato su pattern finanziari realistici per esempi di aggregazione significativi

Connessione a MongoDB

Ora che hai MongoDB e il dataset di esempio pronti, scriviamo il codice per connetterci e verificare il setup:

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())Offerto da

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']}Offerto da

• Verifica: Il codice conferma la connessione e l’accesso al dataset
• Anteprima documenti: Mostra la struttura dei documenti con cui lavorerai

Eseguendo questo codice dovresti vedere un output con il conteggio dei documenti e un’anteprima dei documenti customer e account. Questo conferma che l’ambiente è pronto per gli esempi di pipeline di aggregazione che esploreremo nelle prossime sezioni.

Per un’introduzione a PyMongo e al setup di MongoDB in locale, dai un’occhiata al nostro tutorial PyMongo per principianti.

Approfondimento sulle pipeline di aggregazione MongoDB

Ora che abbiamo confermato la connessione ed esplorato la struttura dei documenti, addentriamoci nel framework di aggregazione di MongoDB. Questa potente funzionalità ci consente di elaborare e trasformare i dati direttamente all’interno del database. In questa sezione esploreremo le fasi di pipeline più comuni con esempi pratici usando il nostro dataset di esempio.

Filtrare i dati con $match

La fase $match filtra i documenti in base a criteri specificati. Pensala come un modo per concentrarti solo sui dati che ti interessano prima di eseguire operazioni più complesse.

Troviamo tutti gli account premium con un limite superiore a $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])Offerto da

Questa pipeline usa la fase $match con l’operatore di confronto $gt (maggiore di) per filtrare gli account. Funziona proprio come il metodo find() ma nel contesto di una pipeline. La query esamina ogni documento nella collection accounts e mantiene solo quelli in cui il campo limit supera 9000.

Output:

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

Osservando i risultati, vediamo che la pipeline ha identificato 1701 account premium su un totale di 1746 account nel nostro database. Questo passaggio di filtraggio restringe il focus, rendendo più efficiente e mirata l’analisi successiva. Nei tuoi progetti potresti usare questa tecnica per concentrarti su utenti attivi, transazioni sopra una certa soglia o prodotti in una categoria specifica prima di approfondire l’analisi di quei documenti.

Rimodellare i documenti con $project, $sort, $limit e $skip

Queste fasi ti aiutano a controllare quali campi includere e come organizzare i risultati. Ecco un esempio che ottiene i 5 account con i limiti più alti, mostrando solo le informazioni essenziali:

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)Offerto da

Questa pipeline ha tre fasi che lavorano insieme:

1. La fase $project rimodella ogni documento:

• Escludendo il campo _id (impostandolo a 0)
• Includendo i campi account_id e limit (impostandoli a 1)
• Creando un nuovo campo chiamato product_count che usa l’operatore $size (uno dei molti operatori della pipeline di aggregazione) per contare gli elementi nell’array products

2. La fase $sort ordina i risultati per il campo limit in ordine decrescente (-1)

3. La fase $limit mantiene solo i primi 5 documenti dopo l’ordinamento

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}Offerto da

L’output mostra i cinque account con i limiti più alti, tutti a $10.000. Possiamo anche vedere quanti prodotti ha ciascun account tramite il nostro campo calcolato product_count. 

Questo consente una presentazione dei dati più pulita e focalizzata su ciò che serve, invece di restituire tutti i campi. Quando costruisci dashboard o report per le tue applicazioni, puoi usare tecniche simili per mostrare solo le informazioni più rilevanti agli utenti, riducendo il trasferimento dati e semplificando la UI.

Raggruppare e aggregare con $group

La fase $group è dove l’aggregazione dà il meglio. Puoi categorizzare i documenti e calcolare metriche per ciascun gruppo. Troviamo la media del limite degli account per tipo di prodotto:

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']}")Offerto da

Questa pipeline usa più operatori per analizzare i dati dei prodotti:

1. L’operatore $unwind suddivide i documenti con più prodotti in documenti separati. Ad esempio, un account con ["Derivatives", "InvestmentStock"] diventa due documenti, uno per ciascun prodotto.
2. La fase $group quindi:

• Raggruppa per nome del prodotto (il campo _id in $group determina la chiave di raggruppamento)
• Calcola il limite medio per ciascun prodotto usando l’accumulatore $avg
• Conta il numero di account per ciascun prodotto usando $sum: 1 (aggiungendo 1 per ogni documento)

3. La fase $sort ordina i risultati per limite medio in ordine decrescente

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: 742Offerto da

I risultati rivelano che i prodotti Commodity sono associati ai limiti medi di account più alti, mentre i prodotti CurrencyService hanno i più bassi. 

Questo tipo di analisi aiuta a identificare correlazioni tra l’offerta di prodotti e la capacità di spesa dei clienti. 

Nelle tue applicazioni potresti usare tecniche simili per analizzare vendite per categoria, engagement degli utenti per funzionalità o errori per modulo—qualsiasi scenario in cui devi riassumere dati su gruppi invece di esaminare singoli record.

Unire collection con $lookup e $unwind

Quando i tuoi dati sono distribuiti su più collection, $lookup ti aiuta a riunirli. Troviamo i clienti con i relativi account:

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'])}")Offerto da

Questa pipeline mostra come unire dati correlati tra collection:

1. La fase $match trova un cliente specifico per username
2. La fase $lookup esegue un left outer join con la collection accounts:

• from: specifica con quale collection effettuare il join
• localField: il campo nella collection corrente (customers) su cui fare match
• foreignField: il campo nella collection di destinazione (accounts) con cui fare match
• as: il nome del nuovo campo array che conterrà i documenti corrispondenti

3. La fase $project modella l’output per mostrare solo i campi rilevanti

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, InvestmentStockOfferto da

L’output fornisce una vista completa del portafoglio finanziario di Elizabeth Ray, mostrando tutti i suoi sei account e i prodotti associati in un unico risultato di query. 

Questo abilita una visione a 360 gradi delle relazioni tra i dati senza richiedere più query o join lato client. 

Per le tue applicazioni, pensa a dove hai suddiviso i dati in più collection per normalizzazione ma hai bisogno di riassemblarli per analisi o visualizzazione. Esempi comuni includono profili utente con cronologia attività, prodotti con stato dell’inventario o contenuti con commenti correlati.

Combinare più fasi per analisi complesse

Ora affrontiamo una domanda di business più articolata: “Quali sono i limiti medi degli account dei clienti, raggruppati per tier di servizio?” Questo richiede di gestire un oggetto annidato all’interno dei documenti customer (tier_and_details) e poi collegare queste informazioni alla collection accounts.

Per prima cosa, per capire la struttura con cui stiamo lavorando nella collection customers, ispezioniamo il campo tier_and_details e il campo accounts per un cliente di esempio. Il campo tier_and_details è un oggetto in cui ogni chiave è un identificatore per un abbonamento a un tier, e il valore contiene dettagli come il nome del tier. Il campo accounts è un array di ID account associati a quel cliente.

# 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']}")Offerto da

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]Offerto da

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']}")Offerto da

Questa pipeline scompone l’analisi complessa in otto passaggi gestibili:

1. $project: Iniziamo trasformando l’oggetto tier_and_details. L’operatore $objectToArray converte questo oggetto in un array di coppie chiave-valore (tiers_array). Questo è essenziale perché fasi come $unwind operano su array. Fondamentale, trasportiamo esplicitamente anche l’array accounts dal documento del cliente come customer_account_ids e l’_id del cliente.
2. $unwind: Questa fase decompone il tiers_array, creando un documento separato per ogni voce di tier che un cliente potrebbe avere. Ogni nuovo documento contiene ancora i customer_account_ids e l’_id originale del cliente.
3. $project: Rimodelliamo il documento per estrarre chiaramente il nome del tier (ad es., "Bronze") dalla struttura annidata ($tiers_array.v.tier) e rinominiamo l’_id del cliente in customer_id per chiarezza. I customer_account_ids vengono mantenuti.
4. $lookup: Qui uniamo con la collection accounts. Usiamo i customer_account_ids (l’array di numeri di account dal documento del cliente) come localField. Il foreignField è account_id della collection accounts. MongoDB troverà tutti gli account il cui account_id è presente nell’array customer_account_ids, aggiungendoli come array nel campo matched_account_details.
5. $unwind: Eseguiamo l’unwind di matched_account_details. Ora, se una combinazione cliente-tier era collegata a più account, otteniamo un documento separato per ciascun account specifico, associato a quel cliente e tier.
6. $group: Raggruppiamo i documenti per tier. Per ogni tier calcoliamo avg_limit usando $avg sul limit dai dettagli account uniti. Contiamo i total_accounts_in_tier usando $sum: 1. Usiamo anche $addToSet con customer_id per raccogliere gli ID cliente unici appartenenti a ciascun tier.
7. $project: La modellazione finale dell’output. Rinominiamo _id (che è il nome del tier dalla fase di group) in tier_name. Usiamo $size per ottenere il conteggio di unique_customers_in_tier.
8. $sort: Ordiniamo i risultati per average_account_limit` in ordine decrescente, così il tier con il limite medio più alto appare per primo.

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: 99Offerto da

I risultati mostrano ora una gerarchia leggermente diversa: i clienti del tier Silver hanno in media account con i limiti più alti, seguiti da Bronze, poi Platinum e infine Gold.

Questo tipo di insight è prezioso per comprendere i segmenti di clientela. Ad esempio, un istituto finanziario potrebbe usare queste informazioni per indagare perché i tier Silver e Bronze hanno limiti medi così alti, o per adattare campagne marketing o offrire servizi premium. 

L’uso di $objectToArray è stato fondamentale per sbloccare i dati dei tier annidati e il passaggio attento dei customer_account_ids ha garantito che il nostro $lookup potesse collegare i clienti ai loro account specifici.

Quando incontri oggetti annidati nei tuoi dataset che devi usare nelle aggregazioni (come preferenze utente, attributi di prodotto o impostazioni di configurazione), ricorda la tecnica $objectToArray. 

Assicurati sempre che tutti i campi necessari per le fasi successive, soprattutto per le operazioni $lookup, siano esplicitamente inclusi nelle tue fasi $project. Questo approccio strutturato alla scomposizione di dati complessi aiuta a ricavare insight significativi direttamente in MongoDB.

Pattern avanzati di pipeline

Oltre alle trasformazioni sequenziali, il framework di aggregazione di MongoDB offre pattern sofisticati per affrontare query analitiche complesse. Quando hai bisogno di analizzare i dati da più angolazioni simultaneamente o di eseguire calcoli basati su un insieme scorrevole di documenti, questi pattern avanzati forniscono soluzioni potenti direttamente nel database.

Analisi multidimensionale con $facet

La fase $facet ti consente di eseguire più sotto-pipeline di aggregazione all’interno di un’unica fase, usando lo stesso insieme di documenti in input. Immagina di dover categorizzare i prodotti per fascia di prezzo e, allo stesso tempo, elencare i 5 brand di prodotto più popolari dalla stessa collection di prodotti. $facet gestisce queste diverse prospettive analitiche in parallelo.

Ogni sotto-pipeline all’interno di $facet opera in modo indipendente sui documenti in input e produce il proprio array di documenti in output. Ciò significa che puoi raccogliere varie metriche o riepiloghi—come quelli necessari per una dashboard completa—tutto in un’unica query al database. 

Per esempio, potresti ottenere il conteggio totale degli utenti attivi, una ripartizione degli utenti per tier di abbonamento e un elenco degli utenti iscritti di recente, tutto dallo stesso dataset utenti in contemporanea. Questo abilita la creazione di report ricchi e multifaccettati senza l’overhead di più chiamate al database, semplificando il recupero dei dati per viste complesse.

Analisi temporale con le funzioni finestra

Introdotte in MongoDB 5.0, le funzioni finestra eseguono calcoli su un insieme di documenti correlati al documento corrente, noto come “finestra”. 

Questo è utile per dati time-series o qualsiasi dataset ordinato in cui è importante il contesto dei documenti vicini. Ad esempio, potresti voler calcolare una media mobile a 7 giorni delle vendite o trovare la somma cumulativa delle transazioni per ciascun cliente nel tempo.

Le funzioni finestra si usano tipicamente all’interno della fase $setWindowFields. Questa fase consente di definire partizioni (gruppi di documenti, come vendite per prodotto) e l’ordine all’interno di tali partizioni (ad esempio per data). 

Poi puoi applicare funzioni finestra come $avg, $sum, $min, $max o funzioni specializzate come $derivative o $integral su una finestra definita (ad es., i 3 documenti precedenti e il documento corrente).

Considera il calcolo di un totale progressivo delle vendite di un prodotto. Una pipeline che usa $setWindowFields potrebbe essere così:

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))Offerto da

In questa pipeline concettuale, per ogni vendita di elettronica viene aggiunto un campo running_total_sales. Questo campo rappresenta la somma di sale_amount per quel prodotto dalla prima vendita fino a quella corrente. 

Calcoli del genere, che in passato richiedevano logiche lato client complesse o più query, possono ora essere eseguiti direttamente nel database, semplificando il codice applicativo e migliorando le prestazioni per analisi di trend o confronti periodo su periodo. 

Conclusione

Le pipeline di aggregazione di MongoDB forniscono un metodo strutturato per elaborare i dati direttamente nel tuo database. Collegando fasi operative distinte, puoi eseguire manipolazioni dei dati articolate, come filtrare record, rimodellare documenti, raggruppare informazioni e combinare dati da varie collection. 

Questo approccio aiuta a rifinire i flussi di lavoro sui dati, consentendo di estrarre più rapidamente significato da grandi dataset, come mostrato negli esempi con PyMongo. Elaborare i dati vicino alla loro origine può spesso portare a prestazioni migliori per le query analitiche.

Sviluppare dimestichezza con le pipeline di aggregazione ti prepara a molte sfide di analisi dei dati. Per mettere in pratica ciò che hai imparato, puoi provare un progetto pratico come Building a Retail Data Pipeline. 

Per una prospettiva più ampia sui sistemi di dati e su come le competenze MongoDB si inseriscono nel quadro generale, il percorso Data Engineer with Python offre un apprendimento approfondito. Se punti alla crescita professionale, rivedere le domande di colloquio su MongoDB può anche essere utile per capire le aspettative tipiche nella risoluzione dei problemi.