Aggregation dengan Python

Pertanyaan bisnis seperti “berapa omzet tiap kategori buku minggu ini?” tidak bisa dijawab satu query find() — data harus difilter, diurai, digabung, lalu dijumlahkan. Aggregation pipeline adalah alat untuk pekerjaan itu, dan di PyMongo bentuknya sangat alami: pipeline adalah list yang berisi dict per stage, persis seperti JSON yang kita tulis di mongosh. Chapter ini menjalankan dua pipeline sungguhan untuk toko buku kita — ringkasan katalog, lalu laporan omzet yang menggabungkan dua collection lewat $lookup.

Pipeline sebagai List of Dict #

Satu berkas berisi katalog, data penjualan, dan dua pipeline. Jalankan utuh, lalu bandingkan hasilnya dengan isi stage-nya:

import os
from datetime import datetime, timedelta, timezone

from pymongo import MongoClient

URI = os.environ.get(
    "MONGODB_URI", "mongodb://127.0.0.1:27017/?directConnection=true"
)

client = MongoClient(URI, serverSelectionTimeoutMS=5000)
client.drop_database("latihan_agregasi_py")
db = client["latihan_agregasi_py"]

db.buku.insert_many([
    {"judul": "Laut Bercerita", "kategori": "fiksi", "harga": 98000},
    {"judul": "Bumi", "kategori": "fiksi", "harga": 89000},
    {"judul": "Filosofi Teras", "kategori": "nonfiksi", "harga": 89000},
    {"judul": "Sapiens", "kategori": "nonfiksi", "harga": 135000},
    {"judul": "Kamus Besar Bahasa Indonesia", "kategori": "referensi",
     "harga": 75000},
])

sejak = datetime.now(timezone.utc) - timedelta(days=7)
db.penjualan.insert_many([
    {"tanggal": sejak, "kasir": "Dewi", "items": [
        {"judul": "Laut Bercerita", "qty": 2, "harga_satuan": 98000},
        {"judul": "Filosofi Teras", "qty": 1, "harga_satuan": 89000}]},
    {"tanggal": sejak, "kasir": "Rina", "items": [
        {"judul": "Bumi", "qty": 1, "harga_satuan": 89000}]},
    {"tanggal": sejak, "kasir": "Dewi", "items": [
        {"judul": "Sapiens", "qty": 3, "harga_satuan": 135000},
        {"judul": "Kamus Besar Bahasa Indonesia", "qty": 2, "harga_satuan": 75000}]},
])

# Pipeline 1: ringkasan katalog — $group lalu $sort
pipeline_katalog = [
    {"$group": {"_id": "$kategori",
                "jumlah_judul": {"$sum": 1},
                "harga_rata": {"$avg": "$harga"}}},
    {"$sort": {"jumlah_judul": -1, "_id": 1}},
]
print("== Ringkasan katalog per kategori ==")
kursor = db.buku.aggregate(pipeline_katalog)
print("tipe hasil aggregate():", type(kursor).__name__)
for baris in kursor:
    print(f"  {baris['_id']:10}: {baris['jumlah_judul']} judul, "
          f"harga rata-rata {baris['harga_rata']}")

# Pipeline 2: omzet per kategori, dengan $lookup ke koleksi buku
pipeline_omzet = [
    {"$match": {"tanggal": {"$gte": sejak}}},
    {"$unwind": "$items"},
    {"$lookup": {"from": "buku", "localField": "items.judul",
                 "foreignField": "judul", "as": "info_buku"}},
    {"$unwind": "$info_buku"},
    {"$group": {"_id": "$info_buku.kategori",
                "omzet": {"$sum": {"$multiply": ["$items.qty",
                                                  "$items.harga_satuan"]}},
                "terjual": {"$sum": "$items.qty"}}},
    {"$sort": {"omzet": -1}},
]
print("== Omzet penjualan per kategori ==")
hasil = list(db.penjualan.aggregate(pipeline_omzet, allowDiskUse=True))
for baris in hasil:
    print(f"  {baris['_id']:10}: omzet {baris['omzet']}, "
          f"terjual {baris['terjual']}")

client.drop_database("latihan_agregasi_py")
client.close()

Output yang Dihasilkan:

== Ringkasan katalog per kategori ==
tipe hasil aggregate(): CommandCursor
  fiksi     : 2 judul, harga rata-rata 93500.0
  nonfiksi  : 2 judul, harga rata-rata 112000.0
  referensi : 1 judul, harga rata-rata 75000.0
== Omzet penjualan per kategori ==
  nonfiksi  : omzet 494000, terjual 4
  fiksi     : omzet 285000, terjual 3
  referensi : omzet 150000, terjual 2

Mari kita bedah bagian pertama:

1. pipeline_katalog = [ ... ] — pipeline adalah list; setiap elemennya satu stage berbentuk dict dengan satu kunci $namaStage. Urutan list = urutan eksekusi di server.
2. $group — field _id menentukan kunci pengelompokan ("$kategori" berarti nilai field kategori), dan akumulator di sebelahnya menghitung nilai per kelompok: $sum: 1 menghitung anggota, $avg merata-ratakan harga.
3. $sort — memakai bentuk yang sama dengan sort pada find(); {"jumlah_judul": -1, "_id": 1} mengurutkan jumlah menurun lalu nama naik sebagai pemecah seri.
4. db.buku.aggregate(pipeline) — mengembalikan CommandCursor, bukan list. Ia diiterasi persis seperti cursor find(); bila butuh list penuh sekaligus, bungkus dengan list() seperti pada pipeline kedua.

$unwind dan $lookup #

Pipeline kedua menjahit dua collection. Setiap document penjualan menyimpan array items, dan kategori hanya ada di collection buku — inilah perjalanan enam stage-nya:

Dua detail layak digarisbawahi. Pertama, $lookup selalu menghasilkan array bernama sesuai as — itulah mengapa $unwind kedua diperlukan sebelum $info_buku.kategori bisa dipakai sebagai kunci group. Kedua, $match diletakkan di posisi paling depan karena data yang tersaring lebih awal membuat stage berikutnya mengerjakan lebih sedikit; prinsip early filtering ini berlaku di semua pipeline.

Satu perbedaan praktis dengan find() juga perlu diingat: cursor hasil aggregate() adalah CommandCursor, bukan Cursor. Method rantai seperti .sort(), .skip(), dan .limit() tidak tersedia di sana — pengurutan dan pembatasan wajib ditulis sebagai stage $sort, $skip, dan $limit di dalam pipeline. Cursor agregasi hanya tahu satu cara dibaca: diiterasi dari awal hingga akhir.

allowDiskUse untuk Data Besar #

Setiap stage aggregation dibatasi sekitar 100 MB memori di server — pipeline yang melebihi batas itu gagal dengan error, kecuali kita mengizinkan server menumpahkan data sementara ke disk. Itulah peran argumen allowDiskUse=True pada contoh kita. Untuk lima document latihan argumen ini jelas tak terpakai, tetapi membiasakan diri menuliskannya (atau menambahkannya saat error batas memori muncul) menyelamatkan pipeline produksi yang tumbuh besar seiring datanya.

Rangkuman #

• Pipeline adalah list of dict — satu elemen per stage, urutan list adalah urutan eksekusi.
• aggregate() mengembalikan CommandCursor; iterasi langsung atau materialisasi dengan list().
• $match di depan memangkas data sedini mungkin; $unwind meratakan array; $lookup menggabungkan collection lain dan hasilnya berupa array.
• allowDiskUse=True mengizinkan stage menumpah ke disk ketika data melampaui batas memori 100 MB.
• Semantik stage identik dengan mongosh — hanya wadahnya yang berubah menjadi struktur data Python.

Aggregation beres. Chapter berikutnya menyorot objek yang selama ini diam-diam kita pakai: cursor — bagaimana ia bekerja malas, dan bagaimana memakainya untuk pagination yang efisien.