Aggregation dengan JavaScript

Laporan penjualan toko buku menuntut perhitungan yang tidak bisa dijawab satu query biasa: omzet per judul, omzet per genre, penyaringan per kanal penjualan. Bisa saja aplikasi menarik seluruh document lalu menjumlahkan dengan loop JavaScript — tetapi pemindahan semua data ke aplikasi hanya untuk dijumlahkan adalah pemborosan bandwidth dan waktu. Aggregation pipeline menempatkan perhitungan di sisi server, dan dari driver Node.js kirimannya berbentuk paling alami di antara semua driver: pipeline hanyalah array of object JavaScript. Chapter ini menyusun dua laporan nyata dari collection penjualan dan buku yang kita tanam sebelumnya.

Pipeline Adalah Array of Stage #

Method aggregate() menerima satu argumen: array berisi stage yang dijalankan server secara berurutan — keluaran satu stage menjadi masukan stage berikutnya. Stage yang dipakai chapter ini:

Karena seluruh stage sudah kita bahas mendalam di Section Aggregation, fokus di sini adalah bentuk pengirimannya dari JavaScript — termasuk bagaimana hasilnya dikonsumsi lewat toArray() pada AggregationCursor.

Laporan Penjualan dalam Satu File #

Data kita berbentuk strukturi toko sungguhan: collection buku sebagai katalog, dan penjualan sebagai struk belanja ber-array item. Laporan pertama menjawab “omzet kanal daring per judul”, laporan kedua menjawab “omzet per genre” dengan menggabungkan kedua collection lewat $lookup:

// laporan-penjualan.js — aggregation pipeline dari driver Node.js
// Jalankan: MONGODB_URI="mongodb://127.0.0.1:27017/?directConnection=true" node laporan-penjualan.js
const { MongoClient } = require("mongodb")

const uri = process.env.MONGODB_URI ?? "mongodb://127.0.0.1:27017/?directConnection=true"

async function main() {
  const client = new MongoClient(uri)
  await client.connect()
  const db = client.db("latihan_js")
  const buku = db.collection("buku")
  const penjualan = db.collection("penjualan")
  await buku.drop().catch(() => {})
  await penjualan.drop().catch(() => {})

  await buku.insertMany([
    { judul: "Laskar Pelangi", genre: "fiksi", harga: 95000, stok: 12 },
    { judul: "Bumi Manusia", genre: "fiksi", harga: 110000, stok: 8 },
    { judul: "Atomic Habits", genre: "pengembangan diri", harga: 125000, stok: 20 },
    { judul: "Filosofi Teras", genre: "pengembangan diri", harga: 90000, stok: 15 },
    { judul: "Cantik Itu Luka", genre: "fiksi", harga: 105000, stok: 6 },
  ])

  await penjualan.insertMany([
    { kanal: "daring", item: [ { judul: "Laskar Pelangi", qty: 2, harga: 95000 }, { judul: "Atomic Habits", qty: 1, harga: 125000 } ] },
    { kanal: "daring", item: [ { judul: "Bumi Manusia", qty: 1, harga: 110000 }, { judul: "Filosofi Teras", qty: 2, harga: 90000 } ] },
    { kanal: "toko", item: [ { judul: "Cantik Itu Luka", qty: 3, harga: 105000 }, { judul: "Laskar Pelangi", qty: 1, harga: 95000 } ] },
  ])

  // Pipeline 1: omzet kanal daring per judul — $match, $unwind, $group, $sort
  const perJudul = await penjualan.aggregate([
    { $match: { kanal: "daring" } },
    { $unwind: "$item" },
    { $group: {
      _id: "$item.judul",
      qty: { $sum: "$item.qty" },
      omzet: { $sum: { $multiply: ["$item.qty", "$item.harga"] } },
    } },
    { $sort: { omzet: -1 } },
  ]).toArray()
  console.log("omzet daring per judul:")
  console.log(perJudul)

  // Pipeline 2: omzet per genre — $lookup ke collection buku + allowDiskUse
  const perGenre = await penjualan.aggregate([
    { $unwind: "$item" },
    { $lookup: { from: "buku", localField: "item.judul", foreignField: "judul", as: "infoBuku" } },
    { $unwind: "$infoBuku" },
    { $group: {
      _id: "$infoBuku.genre",
      omzet: { $sum: { $multiply: ["$item.qty", "$item.harga"] } },
    } },
    { $sort: { omzet: -1 } },
  ], { allowDiskUse: true }).toArray()
  console.log("omzet per genre (semua kanal, via $lookup):")
  console.log(perGenre)

  await penjualan.drop()
  await buku.drop()
  await client.close()
}

main().catch(err => {
  console.error("GAGAL:", err.name, "-", err.message)
  process.exit(1)
})

MONGODB_URI="mongodb://127.0.0.1:27017/?directConnection=true" node laporan-penjualan.js

Output yang Dihasilkan:

omzet daring per judul:
[
  { _id: 'Laskar Pelangi', qty: 2, omzet: 190000 },
  { _id: 'Filosofi Teras', qty: 2, omzet: 180000 },
  { _id: 'Atomic Habits', qty: 1, omzet: 125000 },
  { _id: 'Bumi Manusia', qty: 1, omzet: 110000 }
]
omzet per genre (semua kanal, via $lookup):
[
  { _id: 'fiksi', omzet: 710000 },
  { _id: 'pengembangan diri', omzet: 305000 }
]

Mari kita bedah jalannya pipeline pertama: $match: { kanal: "daring" } membuang struk kanal toko lebih dulu agar stage berikutnya bekerja lebih ringan; $unwind: "$item" mengubah setiap struk dua item menjadi dua document satu item — inilah yang membuat $group bisa menjumlahkan per judul; $group dengan _id: "$item.judul" mengelompokkan pecahan tadi, sementara omzet menjumlahkan hasil $multiply qty dan harga per document; $sort: { omzet: -1 } menutup dengan urutan omzet menurun.

Pipeline kedua memakai pola join: $lookup mencocokkan item.judul pada struk dengan field judul di collection buku, lalu hasilnya diletakkan sebagai array infoBuku — $unwind: "$infoBuku" membukanya menjadi field biasa sehingga $group dapat mengelompokkan berdasarkan $infoBuku.genre. Angka fiksi (710000) berasal dari tiga judul: Laskar Pelangi 285000, Bumi Manusia 110000, dan Cantik Itu Luka 315000.

Tiga detail kecil pada $lookup menentukan hasilnya: from menyebut collection tujuan (buku), pasangan localField/foreignField adalah kunci penghubung dua sisi, dan as memberi nama pada array hasil. Cocok satu-struk-banyak-buku membuat array itu berisi paling banyak satu elemen — satu judul hanya muncul sekali di katalog — sehingga $unwind berikutnya aman; bila kunci bisa cocok berkali-kali, array membesar dan perlu difilter sebelum dikelompokkan.

allowDiskUse Saat Stage Kehabisan Memori #

Secara bawaan, setiap stage aggregation dibatasi memori 100 MB — batasnya per stage, bukan per pipeline. Data latihan kita jauh di bawah itu, tetapi pipeline produksi yang mengelompokkan jutaan document dapat menemui batas tersebut — dan server menolaknya dengan galat. Opsi kedua aggregate(pipeline, { allowDiskUse: true }) seperti pada pipeline kedua di atas memberi izin stage seperti $sort dan $group menumpahkan data sementara ke disk agar tetap selesai. Stage yang paling sering menabrak batas ini adalah $group dengan kardinalitas tinggi dan $sort pada kumpulan besar.

Rangkuman #

• aggregate() menerima array stage; dari JavaScript bentuknya persis pipeline mongosh — $match, $unwind, $group, $sort, $lookup semuanya object biasa.
• $unwind memecah document ber-array sebelum dikelompokkan; $lookup menggabungkan collection lain lalu biasanya diikuti $unwind hasilnya.
• Hasil berupa AggregationCursor — toArray() memuat semuanya, atau diiterasi bertahap seperti Chapter berikutnya.
• { allowDiskUse: true } mengizinkan stage berat menulis ke disk saat melewati batas memori 100 MB per stage.