Tutorial Pipeline Agregasi MongoDB di Python dengan PyMongo

Bekerja dengan dataset besar sering menghadirkan tantangan untuk mengekstrak pola dan wawasan bermakna sekaligus menjaga kinerja. Saat aplikasi Anda menyimpan data di MongoDB, menjalankan kueri dan transformasi kompleks langsung di database bisa jauh lebih cepat daripada memindahkan data ke alat analisis eksternal. Pipeline agregasi MongoDB memberikan solusi dengan memungkinkan Anda memproses, mentransformasi, dan menganalisis data tepat di tempatnya.

Anda dapat membuat alur kerja pemrosesan data khusus dengan menghubungkan operasi sederhana secara berurutan. Setiap tahap dalam pipeline mentransformasi dokumen dan meneruskan hasilnya ke tahap berikutnya. Misalnya, Anda mungkin perlu memfilter catatan berdasarkan rentang tanggal, mengelompokkannya menurut kategori, menghitung ukuran statistik, dan memformat output—semuanya diselesaikan melalui satu operasi database yang memproses data sedekat mungkin dengan sumbernya.

Dalam artikel ini, Anda akan mempelajari cara membangun pipeline agregasi untuk menyelesaikan tantangan data umum, dengan semua contoh ditunjukkan dalam PyMongo (klien Python resmi MongoDB). 

Walaupun artikel ini berfokus pada pipeline agregasi, jika Anda baru menggunakan MongoDB dengan Python, kursus Introduction to MongoDB in Python menawarkan titik awal yang komprehensif. Anda akan mendapatkan fondasi yang cukup dari artikel ini untuk menerapkan konsep agregasi tersebut ke bahasa kueri MongoDB sendiri atau dengan bantuan model bahasa.

Apa itu Pipeline Agregasi di MongoDB?

Bayangkan Anda perlu menganalisis ulasan pelanggan di berbagai produk untuk memahami tren kepuasan. Kueri tradisional mungkin mengambil datanya, tetapi tidak membantu dalam menggabungkan, menganalisis, dan mentransformasi informasi tersebut menjadi ringkasan yang berguna. 

Pipeline agregasi MongoDB memecahkan ini dengan menyediakan cara terstruktur untuk memproses data melalui rangkaian operasi yang saling membangun.

Arsitektur pipeline dan alur dokumen

Anggap pipeline agregasi sebagai lini perakitan untuk data Anda. Setiap dokumen dari koleksi Anda masuk ke salah satu ujung pipeline dan melewati berbagai stasiun (tahap) tempat dokumen tersebut difilter, ditransformasi, dikelompokkan, atau diperkaya. 

Output dari satu tahap menjadi input untuk tahap berikutnya, memungkinkan Anda memecah transformasi data yang kompleks menjadi langkah-langkah kecil yang dapat dikelola.

Pipeline ini menggunakan pendekatan deklaratif; Anda menentukan apa yang diinginkan di setiap tahap, bukan bagaimana cara menghitungnya. Pendekatan ini membuat niat pemrosesan data Anda jelas dan memungkinkan MongoDB menangani detail eksekusinya. Database kemudian dapat menerapkan berbagai optimasi berdasarkan struktur pipeline Anda.

Urutan tahap penting dalam desain pipeline Anda. Memfilter dokumen lebih awal (sebelum pengelompokan atau perhitungan kompleks) mengurangi jumlah data yang mengalir melalui pipeline. 

Pendekatan ini dapat secara dramatis meningkatkan kinerja saat bekerja dengan koleksi besar. Pipeline yang tertata baik hanya memproses data yang diperlukan untuk hasil akhir Anda.

• Alur dokumen: Dokumen masuk → ditransformasi melalui tahap → hasil keluar
• Desain deklaratif: Tentukan apa yang Anda inginkan, bukan cara menghitungnya
• Pemrosesan berurutan: Output setiap tahap menjadi input langsung untuk tahap berikutnya
• Manfaat kinerja: Penyaringan awal mengurangi volume data di tahap selanjutnya

Tipologi tahap dan kategori fungsional

Tahap agregasi MongoDB terbagi dalam empat kategori utama berdasarkan tujuannya. Tahap penyaringan seperti $match bekerja seperti kueri, memilih hanya dokumen yang memenuhi kriteria tertentu. Ini membantu mempersempit dataset Anda sebelum melakukan operasi yang lebih kompleks.

Tahap pembentukan ulang mentransformasi struktur dokumen. Dengan $project atau $addFields, Anda dapat menyertakan, mengecualikan, atau mengganti nama field, atau membuat field terhitung berdasarkan nilai yang ada. Tahap-tahap ini membantu menyederhanakan dokumen dengan hanya mempertahankan informasi relevan dan menambahkan nilai terhitung yang diperlukan untuk analisis.

Saat Anda perlu menggabungkan beberapa dokumen berdasarkan karakteristik bersama, tahap pengelompokan berperan. Tahap $group adalah andalannya, memungkinkan Anda menghitung jumlah, total, rata-rata, dan nilai agregat lain di seluruh kelompok dokumen. Ini mengubah ribuan catatan individual menjadi ringkasan bermakna yang menjawab pertanyaan analitis Anda.

Untuk melengkapi gambaran data Anda, tahap penggabungan seperti $lookup memungkinkan Anda menggabungkan informasi dari beberapa koleksi. Ini memungkinkan Anda memperkaya dokumen dengan data terkait, mirip dengan join SQL tetapi disesuaikan untuk model dokumen MongoDB. Kemampuan merujuk data lintas koleksi membantu menjaga normalisasi data yang tepat sekaligus memberikan hasil lengkap dalam satu operasi.

• Penyaringan: Pilih dokumen tertentu dengan $match berdasarkan kriteria
• Pembentukan ulang: Transformasi struktur dokumen dengan $project dan $addFields
• Pengelompokan: Gabungkan dan rangkum dengan $group untuk nilai agregat
• Penggabungan: Hubungkan data terkait lintas koleksi dengan $lookup

Untuk referensi lengkap tentang kapabilitas dan operator, Anda dapat merujuk ke MongoDB Aggregation Pipeline Manual.

Menyiapkan Lingkungan untuk Tutorial

Sebelum mendalami pipeline agregasi MongoDB, Anda memerlukan lingkungan kerja dengan PyMongo dan akses ke dataset contoh. Bagian ini memandu Anda melalui proses penyiapan dengan dataset sample_analytics, yang berisi data keuangan yang cocok untuk mendemonstrasikan konsep agregasi.

Menginstal PyMongo

PyMongo adalah driver Python resmi untuk MongoDB. Anda dapat menginstalnya menggunakan pip di macOS maupun Windows:

# Install PyMongo using pip
pip install pymongo

# Or if you're using conda
conda install -c conda-forge pymongoDidukung oleh

• Kesesuaian versi: PyMongo 4.x bekerja dengan Python 3.7+
• Penanganan dependensi: Installer secara otomatis menangani dependensi yang diperlukan
• Dukungan platform: Bekerja identik di Windows, macOS, dan Linux

Mengakses dataset contoh

MongoDB menawarkan dataset contoh yang dapat Anda gunakan tanpa membuat data sendiri. Cara termudah adalah menggunakan MongoDB Atlas (versi cloud):

1. Buat akun MongoDB Atlas gratis di mongodb.com/cloud/atlas
2. Siapkan cluster tier gratis

2.1. Pilih versi free forever

2.2. Beri nama cluster Anda

2.3. Klik "Deploy"

2.4. Salin nama pengguna dan kata sandi Anda ke tempat yang aman

2.5. Klik "Create database user"

2.6. Pilih "Drivers" untuk metode koneksi Anda

2.7. Salin URI koneksi untuk langkah berikutnya

3.1. Klik tiga titik di tampilan cluster Anda

3.2. Pilih "Load sample dataset"

3.3. Dari daftar, pilih sample_analytics dan tunggu hingga selesai dimuat.

Untuk instalasi MongoDB lokal, Anda dapat memuat dataset contoh menggunakan:

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

• Ukuran data: Dataset sample_analytics cukup kecil (~10MB) untuk berjalan baik di tier gratis
• Struktur koleksi: Berisi koleksi customers dan accounts dengan data relasi
• Realistisnya data: Berdasarkan pola keuangan realistis untuk contoh agregasi yang bermakna

Menyambungkan ke MongoDB

Sekarang Anda sudah menyiapkan MongoDB dan dataset contoh, mari tulis kode untuk menyambung dan memverifikasi penyiapan kita:

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())Didukung oleh

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']}Didukung oleh

• Verifikasi: Kode mengonfirmasi koneksi dan akses dataset Anda
• Pratinjau dokumen: Menampilkan struktur dokumen yang akan Anda gunakan

Saat Anda menjalankan kode ini, Anda akan melihat output yang menampilkan jumlah dokumen dan pratinjau dokumen pelanggan serta akun. Ini mengonfirmasi bahwa lingkungan Anda siap untuk contoh pipeline agregasi yang akan kita jelajahi di bagian berikutnya.

Untuk pengantar PyMongo dan penyiapan MongoDB lokal, lihat tutorial PyMongo untuk pemula.

Mendalami Pipeline Agregasi MongoDB

Sekarang kita telah mengonfirmasi koneksi dan menelusuri struktur dokumen, mari menyelami kerangka agregasi MongoDB. Fitur kuat ini memungkinkan kita memproses dan mentransformasi data langsung di dalam database. Di bagian ini, kita akan mengeksplorasi tahap pipeline paling umum dengan contoh praktis menggunakan dataset contoh kita.

Memfilter data dengan $match

Tahap $match memfilter dokumen berdasarkan kriteria tertentu. Anggap ini sebagai cara untuk fokus hanya pada data yang Anda butuhkan sebelum melakukan operasi yang lebih kompleks.

Mari temukan semua akun premium dengan limit di atas $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])Didukung oleh

Pipeline ini menggunakan tahap $match dengan operator perbandingan $gt (lebih besar dari) untuk memfilter akun. Cara kerjanya mirip dengan metode find() tetapi dalam konteks pipeline. Kueri memeriksa setiap dokumen di koleksi accounts dan hanya mempertahankan dokumen yang field limit-nya melebihi 9000.

Output:

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

Melihat hasilnya, kita dapat melihat pipeline mengidentifikasi 1701 akun premium dari total 1746 akun di database kita. Langkah penyaringan ini mempersempit fokus, membuat analisis selanjutnya lebih efisien dan terarah. Dalam proyek Anda sendiri, Anda dapat menggunakan teknik ini untuk fokus pada pengguna aktif, transaksi di atas jumlah tertentu, atau produk dalam kategori spesifik sebelum melakukan analisis lebih mendalam pada dokumen tersebut.

Membentuk ulang dokumen dengan $project, $sort, $limit, dan $skip

Tahap-tahap ini membantu Anda mengontrol field mana yang akan disertakan dan bagaimana mengatur hasilnya. Berikut contoh yang mengambil 5 akun teratas dengan limit tertinggi, hanya menampilkan informasi esensial:

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)Didukung oleh

Pipeline ini memiliki tiga tahap yang bekerja bersama:

1. Tahap $project membentuk ulang setiap dokumen dengan:

• Mengecualikan field _id (diatur ke 0)
• Menyertakan field account_id dan limit (diatur ke 1)
• Membuat field baru bernama product_count yang menggunakan operator $size (salah satu dari banyak operator pipeline agregasi) untuk menghitung item dalam array products

2. Tahap $sort mengurutkan hasil berdasarkan field limit secara menurun (-1)

3. Tahap $limit mempertahankan hanya 5 dokumen pertama setelah pengurutan

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}Didukung oleh

Output menunjukkan lima akun dengan limit tertinggi, semuanya di $10.000. Kita juga dapat melihat berapa banyak produk yang dimiliki setiap akun melalui field product_count yang dihitung.

Ini memungkinkan presentasi data yang lebih ringkas dan fokus pada kebutuhan, alih-alih mengembalikan semua field. Saat membangun dasbor atau laporan untuk aplikasi Anda, Anda dapat menggunakan teknik serupa untuk menampilkan hanya informasi paling relevan bagi pengguna, mengurangi transfer data dan menyederhanakan UI.

Mengelompokkan dan mengagregasi dengan $group

Tahap $group adalah tempat agregasi benar-benar bersinar. Anda dapat mengategorikan dokumen dan menghitung metrik di setiap kelompok. Mari cari rata-rata limit akun berdasarkan jenis produk:

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']}")Didukung oleh

Pipeline ini menggunakan beberapa operator untuk menganalisis data produk:

1. Operator $unwind membagi dokumen dengan banyak produk menjadi dokumen terpisah. Misalnya, sebuah akun dengan ["Derivatives", "InvestmentStock"] menjadi dua dokumen, satu untuk setiap produk.
2. Tahap $group kemudian:

• Mengelompokkan berdasarkan nama produk (field _id di $group menentukan kunci pengelompokan)
• Menghitung rata-rata limit untuk setiap produk menggunakan akumulator $avg
• Menghitung jumlah akun untuk setiap produk menggunakan $sum: 1 (menambah 1 untuk setiap dokumen)

3. Tahap $sort mengurutkan hasil berdasarkan rata-rata limit secara menurun

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: 742Didukung oleh

Hasilnya mengungkap bahwa produk Commodity terkait dengan rata-rata limit akun tertinggi, sementara produk CurrencyService memiliki yang terendah. 

Analisis seperti ini membantu mengidentifikasi korelasi antara penawaran produk dan kapasitas belanja pelanggan. 

Dalam aplikasi Anda sendiri, Anda dapat menggunakan teknik serupa untuk menganalisis penjualan per kategori, keterlibatan pengguna per fitur, atau error per modul—skenario apa pun di mana Anda perlu merangkum data lintas kelompok alih-alih meneliti catatan individual.

Melakukan join antarkoleksi dengan $lookup dan $unwind

Saat data Anda tersebar di beberapa koleksi, $lookup membantu menyatukannya. Mari temukan pelanggan beserta akun terkaitnya:

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'])}")Didukung oleh

Pipeline ini mendemonstrasikan cara melakukan join data terkait lintas koleksi:

1. Tahap $match menemukan pelanggan tertentu berdasarkan username
2. Tahap $lookup melakukan left outer join dengan koleksi accounts:

• from: menentukan koleksi mana yang akan di-join
• localField: field di koleksi saat ini (customers) untuk dicocokkan
• foreignField: field di koleksi target (accounts) untuk dicocokkan
• as: nama untuk field array baru yang akan berisi dokumen yang cocok

3. Tahap $project membentuk output untuk menampilkan hanya field yang relevan

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, InvestmentStockDidukung oleh

Output memberikan gambaran lengkap portofolio finansial Elizabeth Ray, menampilkan keenam akunnya dan produk terkait dalam satu hasil kueri.

Ini memungkinkan pandangan 360 derajat atas relasi data Anda tanpa memerlukan banyak kueri atau join di sisi klien.

Untuk aplikasi Anda sendiri, pertimbangkan di mana Anda memisahkan data lintas koleksi untuk tujuan normalisasi namun perlu merakitnya kembali untuk analisis atau tampilan. Contoh umum termasuk profil pengguna dengan riwayat aktivitas, produk dengan status inventaris, atau konten dengan komentar terkait.

Menggabungkan banyak tahap untuk analisis kompleks

Sekarang, mari kita tangani pertanyaan bisnis yang lebih mendalam: “Berapa rata-rata limit akun untuk pelanggan, dikelompokkan menurut tier layanan mereka?” Ini mengharuskan kita menangani objek bertingkat di dalam dokumen pelanggan (tier_and_details) lalu menghubungkan informasi ini ke koleksi accounts.

Pertama, untuk memahami struktur yang kita hadapi di koleksi customers, mari lihat field tier_and_details dan field accounts untuk satu pelanggan contoh. Field tier_and_details adalah objek di mana setiap kunci adalah pengenal untuk langganan tier, dan nilainya memuat detail seperti nama tier. Field accounts adalah array ID akun yang terkait dengan pelanggan tersebut.

# 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']}")Didukung oleh

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]Didukung oleh

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']}")Didukung oleh

Pipeline ini memecah analisis kompleks menjadi delapan langkah yang dapat ditangani:

1. $project: Kita mulai dengan mentransformasi objek tier_and_details. Operator $objectToArray mengonversi objek ini menjadi array pasangan key-value (tiers_array). Ini penting karena tahap seperti $unwind beroperasi pada array. Yang krusial, kita juga secara eksplisit meneruskan array accounts dari dokumen pelanggan sebagai customer_account_ids dan _id pelanggan.
2. $unwind: Tahap ini membongkar tiers_array, membuat dokumen terpisah untuk setiap entri tier yang mungkin dimiliki pelanggan. Setiap dokumen baru masih memuat customer_account_ids dan _id pelanggan asli.
3. $project: Kita membentuk ulang dokumen untuk dengan jelas mengekstrak nama tier (mis., "Bronze") dari struktur bertingkat ($tiers_array.v.tier) dan mengganti nama _id pelanggan menjadi customer_id untuk kejelasan. customer_account_ids diteruskan.
4. $lookup: Di sini kita melakukan join dengan koleksi accounts. Kita menggunakan customer_account_ids (array nomor akun dari dokumen pelanggan) sebagai localField. foreignField adalah account_id dari koleksi accounts. MongoDB akan menemukan semua akun yang account_id-nya ada di array customer_account_ids, lalu menambahkannya sebagai array ke field matched_account_details.
5. $unwind: Kita melakukan unwind pada matched_account_details. Sekarang, jika kombinasi pelanggan-tier terhubung ke beberapa akun, kita mendapatkan dokumen terpisah untuk setiap akun spesifik yang terkait dengan pelanggan dan tier tersebut.
6. $group: Kita mengelompokkan dokumen berdasarkan tier. Untuk setiap tier, kita menghitung avg_limit menggunakan $avg pada limit dari detail akun yang di-join. Kita menghitung total_accounts_in_tier menggunakan $sum: 1. Kita juga menggunakan $addToSet dengan customer_id untuk mengumpulkan ID pelanggan unik yang termasuk di setiap tier.
7. $project: Pembentukan akhir output kita. Kita mengganti nama _id (yang merupakan nama tier dari tahap pengelompokan) menjadi tier_name. Kita menggunakan $size untuk mendapatkan jumlah unique_customers_in_tier.
8. $sort: Kita mengurutkan hasil berdasarkan average_account_limit` secara menurun, sehingga tier dengan rata-rata limit tertinggi muncul pertama.

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: 99Didukung oleh

Hasilnya kini menunjukkan hierarki yang sedikit berbeda: pelanggan tier Silver rata-rata memiliki akun dengan limit tertinggi, diikuti oleh Bronze, lalu Platinum, dan terakhir Gold.

Wawasan seperti ini berharga untuk memahami segmen pelanggan. Misalnya, sebuah institusi keuangan dapat menggunakan informasi ini untuk menyelidiki mengapa tier Silver dan Bronze memiliki rata-rata limit begitu tinggi, atau untuk menyesuaikan kampanye pemasaran atau menawarkan layanan premium.

Penggunaan $objectToArray menjadi kunci untuk membuka data tier bertingkat, dan dengan cermat meneruskan customer_account_ids memastikan $lookup dapat menghubungkan pelanggan ke akun spesifik mereka.

Saat Anda menemui objek bertingkat dalam dataset Anda yang perlu digunakan dalam agregasi (seperti preferensi pengguna, atribut produk, atau pengaturan konfigurasi), ingat teknik $objectToArray. 

Selalu pastikan bahwa field apa pun yang diperlukan untuk tahap selanjutnya, terutama untuk operasi $lookup, secara eksplisit disertakan dalam tahap $project Anda. Pendekatan terstruktur untuk memecah data kompleks ini membantu memperoleh wawasan bermakna langsung di dalam MongoDB.

Pola Pipeline Lanjutan

Di luar transformasi berurutan, kerangka agregasi MongoDB menawarkan pola canggih untuk menangani kueri analitis yang kompleks. Saat Anda perlu menganalisis data dari berbagai sudut secara bersamaan atau melakukan perhitungan berdasarkan kumpulan dokumen yang bergerak, pola lanjutan ini menyediakan solusi kuat langsung di dalam database.

Analitik multidimensi dengan $facet

Tahap $facet memungkinkan Anda menjalankan beberapa sub-pipeline agregasi dalam satu tahap, menggunakan kumpulan dokumen masukan yang sama. Bayangkan perlu mengategorikan produk berdasarkan rentang harga dan, pada saat yang sama, membuat daftar 5 merek produk paling populer dari koleksi produk yang sama. $facet menangani berbagai perspektif analitis ini secara paralel.

Setiap sub-pipeline dalam $facet beroperasi secara independen pada dokumen masukan dan menghasilkan array outputnya sendiri. Ini berarti Anda dapat mengumpulkan berbagai metrik atau ringkasan—seperti yang diperlukan untuk dasbor komprehensif—semua dalam satu kueri database. 

Sebagai contoh, Anda dapat memperoleh jumlah total pengguna aktif, rincian pengguna berdasarkan tier langganan, dan daftar pengguna yang baru bergabung, semuanya dari dataset pengguna yang sama secara bersamaan. Hal ini memungkinkan pembuatan laporan kaya dan multi-faset tanpa overhead banyak panggilan database, sehingga merampingkan pengambilan data untuk tampilan kompleks.

Analisis temporal dengan fungsi window

Diperkenalkan di MongoDB 5.0, fungsi window melakukan perhitungan di atas sekumpulan dokumen yang terkait dengan dokumen saat ini, yang disebut “window.” 

Ini berguna untuk data deret waktu atau dataset berurutan apa pun di mana konteks dari dokumen di sekitarnya penting. Misalnya, Anda mungkin ingin menghitung rata-rata bergerak 7 hari penjualan atau menemukan jumlah kumulatif transaksi untuk setiap pelanggan dari waktu ke waktu.

Fungsi window biasanya digunakan dalam tahap $setWindowFields. Tahap ini memungkinkan Anda mendefinisikan partisi (kelompok dokumen, seperti penjualan per produk) dan urutan pengurutan dalam partisi tersebut (seperti berdasarkan tanggal). 

Kemudian, Anda dapat menerapkan fungsi window seperti $avg, $sum, $min, $max, atau fungsi khusus seperti $derivative atau $integral pada window yang didefinisikan (mis., tiga dokumen sebelumnya dan dokumen saat ini).

Pertimbangkan menghitung total berjalan penjualan produk. Pipeline menggunakan $setWindowFields mungkin terlihat seperti ini:

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))Didukung oleh

Dalam pipeline konseptual ini, untuk setiap penjualan elektronik, field running_total_sales ditambahkan. Field ini merepresentasikan jumlah sale_amount untuk produk tersebut dari penjualan paling awal hingga penjualan saat ini.

Perhitungan seperti ini, yang sebelumnya memerlukan logika sisi klien yang kompleks atau banyak kueri, kini dapat dilakukan langsung di dalam database, menyederhanakan kode aplikasi dan meningkatkan kinerja untuk analisis tren atau perbandingan periode ke periode.

Kesimpulan

Pipeline agregasi MongoDB menyediakan metode terstruktur untuk memproses data langsung di dalam database Anda. Dengan menghubungkan tahap operasional yang berbeda, Anda dapat melakukan manipulasi data yang kompleks, seperti memfilter catatan, membentuk ulang dokumen, mengelompokkan informasi, dan menggabungkan data dari berbagai koleksi.

Pendekatan ini membantu menyempurnakan alur kerja data Anda, memungkinkan ekstraksi makna dari dataset besar dengan lebih cepat, seperti yang ditunjukkan melalui contoh PyMongo. Memproses data sedekat mungkin dengan sumbernya sering kali menghasilkan kinerja yang lebih baik untuk kueri analitis.

Menguasai pipeline agregasi mempersiapkan Anda menghadapi banyak tantangan analisis data. Untuk menerapkan apa yang telah Anda pelajari, Anda dapat mencoba proyek praktik seperti Building a Retail Data Pipeline. 

Untuk perspektif yang lebih luas tentang sistem data dan bagaimana keterampilan MongoDB cocok dalam gambaran besar, track Data Engineer with Python menawarkan pembelajaran mendalam. Jika Anda menatap peningkatan karier, meninjau MongoDB Interview Questions umum juga dapat bermanfaat untuk memahami ekspektasi pemecahan masalah yang tipikal.