aggregation pipeline

find ตอบคำถามได้ชนิดเดียว: “คืน document ที่ตรง filter นี้” ซึ่งพอสำหรับการอ่านเรคคอร์ด แต่ สรุป ไม่ได้ มันบอกไม่ได้ว่าแต่ละ title ขายได้กี่เล่ม ลูกค้าแต่ละคนสั่งกี่ครั้ง หรือหนังสือขายดีสามอันดับแรก สำหรับคำถามที่ยุบ document จำนวนมากเป็นคำตอบที่เล็กและถูกปรับรูป MongoDB มี aggregation pipeline

ลองนึกภาพสายพาน: document เข้าที่ปลายหนึ่ง ผ่านแถวของ stage และสายข้อมูลที่ถูกแปลงออกอีกปลาย แต่ละ stage ทำงานหน้าที่เดียวที่ชัดเจนแล้วส่งผลต่อให้ stage ถัดไป ลำดับเป็นของคุณ และมีผล - stage เห็นได้เฉพาะสิ่งที่ stage ก่อนหน้าผลิต

ชุดข้อมูลเดียวทั้งโมดูล ทุกบทที่นี่เตรียม collection sales เดียวกันในบล็อกแรก: order หนังสือ 200 000 รายการ { title, genre, quantity, price, buyer } จาก 500 title, 5 genre, ราคา 5-25 และผู้ซื้อ 1 000 คน (บท unwind ใช้ collection orders ของตัวเองที่มี array items) output จับจาก mongo 8.3.11, mongosh 2.11.1 และ Node driver 7.7.0 บน Mac เครื่องเดียว; แท็บ Python, Go และ Rust ถูกรันอย่างละครั้งในโปรแกรม scratch aplan() คือ helper ที่ย่อ explain("executionStats") ของ aggregation เป็นบรรทัดเดียว: สิ่งที่ stage แรกอ่านจาก collection แล้วตามด้วย stage ที่ยังเหลือให้รันใน pipeline

mongosh/agg-seed.js

const genres = ["fiction", "science", "history", "poetry", "comics"];
globalThis.aplan = (pipeline) => {               // one line per aggregate explain: what the first stage read, then the stages left in the pipeline
  const e = db.sales.explain("executionStats").aggregate(pipeline);
  const cursor = e.stages ? e.stages[0].$cursor : e, rest = e.stages ? e.stages.slice(1).map((s) => Object.keys(s)[0]) : [];
  const chain = (p) => { const out = []; for (p = p.queryPlan ?? p; p; p = p.inputStage) out.push(p.stage + (p.indexName ? "(" + p.indexName + ")" : "")); return out.join(" > "); };
  const s = cursor.executionStats;
  return chain(cursor.queryPlanner.winningPlan) + "  keys=" + s.totalKeysExamined + " docs=" + s.totalDocsExamined + (rest.length ? "  | left in pipeline: " + rest.join(" > ") : "");
};
for (let s = 0; s < 200000; s += 50000) {
  db.sales.insertMany(Array.from({ length: 50000 }, (_, k) => {
    const i = s + k;                                          // 500 titles, 5 genres, quantity 1..5, price 5..25, 1000 buyers (pseudo-random but repeatable)
    return { title: "T" + (i % 500), genre: genres[(i % 500) % 5], quantity: 1 + ((Math.imul(i, 2654435761) >>> 0) % 5), price: 5 + ((Math.imul(i, 40503) >>> 0) % 21), buyer: "b" + (i % 1000) };
  }), { ordered: false });
}
db.sales.createIndex({ price: 1 });
console.log("sales:", db.sales.countDocuments(), "orders");

sales: 200000 orders

pipeline คือรายการของ stage

คุณส่ง array ของ stage object ให้ aggregation แต่ละ stage ตั้งชื่อด้วย key ที่ขึ้นต้น $ - $match, $group, $sort เป็นต้น รูปแบบมาตรฐาน: กรองสาย พับเป็นกลุ่ม เรียงผลลัพธ์ นี่คือ: สามอันดับ title นิยายที่ขายดีที่สุด

flowchart LR
  In["sales documents"] --> M["$match: filter the stream"]
  M --> G["$group: fold into groups"]
  G --> S["$sort: order the groups"]
  S --> L["$limit: keep three"]
  L --> Out["result documents"]

document ไหลจากซ้ายไปขวา แต่ละ stage แปลงสายข้อมูลและป้อนต่อให้ stage ถัดไป

mongosh

mongosh/agg-first-pipeline.js

const rows = db.sales.aggregate([
  { $match: { genre: "fiction" } },
  { $group: { _id: "$title", copies: { $sum: "$quantity" } } },
  { $sort: { copies: -1, _id: 1 } },
  { $limit: 3 },
]).toArray();
console.log(rows);
assert.strictEqual(rows.length, 3);

Node.js

app/agg-first-pipeline.mjs

const sales = db.collection("sales");
await sales.insertMany([
  { title: "Dune", genre: "fiction", quantity: 3, price: 12, buyer: "Ada" },
  { title: "Dune", genre: "fiction", quantity: 4, price: 12, buyer: "Linus" },
  { title: "Neuromancer", genre: "fiction", quantity: 4, price: 9, buyer: "Grace" },
  { title: "Cosmos", genre: "science", quantity: 5, price: 20, buyer: "Ada" },
]);
const rows = await sales.aggregate([
  { $match: { genre: "fiction" } },
  { $group: { _id: "$title", copies: { $sum: "$quantity" } } },
  { $sort: { copies: -1 } },
]).toArray();
console.log(rows);
if (rows.length !== 2 || rows[0]._id !== "Dune" || rows[0].copies !== 7) throw new Error("unexpected pipeline result");

Python

rows = list(db.sales.aggregate([
    {"$match": {"genre": "fiction"}},
    {"$group": {"_id": "$title", "copies": {"$sum": "$quantity"}}},
    {"$sort": {"copies": -1}},
]))

Go

pipeline := mongo.Pipeline{
    {{Key: "$match", Value: bson.D{{Key: "genre", Value: "fiction"}}}},
    {{Key: "$group", Value: bson.D{
        {Key: "_id", Value: "$title"},
        {Key: "copies", Value: bson.D{{Key: "$sum", Value: "$quantity"}}},
    }}},
    {{Key: "$sort", Value: bson.D{{Key: "copies", Value: -1}}}},
}
cur, err := coll.Aggregate(ctx, pipeline)
if err != nil {
    return err
}
var rows []bson.M
if err := cur.All(ctx, &rows); err != nil {
    return err
}

Rust

let mut cursor = coll
    .aggregate(vec![
        doc! { "$match": { "genre": "fiction" } },
        doc! { "$group": { "_id": "$title", "copies": { "$sum": "$quantity" } } },
        doc! { "$sort": { "copies": -1 } },
    ])
    .await?;

[
  { _id: 'T305', copies: 1205 },
  { _id: 'T345', copies: 1205 },
  { _id: 'T30', copies: 1204 }
]

แท็บ Node.js รันบน collection สี่ order ของตัวเองและคืนหนึ่งแถวต่อ title:

[ { _id: 'Dune', copies: 7 }, { _id: 'Neuromancer', copies: 4 } ]

In Compass: เปิด collection แล้วคลิกแท็บ Aggregations ตัวสร้าง pipeline ให้เพิ่มทีละ stage แก้เนื้อหา และดูตัวอย่าง document ที่ไหลออกจากแต่ละ stage สด ๆ - วิธีที่ดีที่สุดที่จะ ดู สายพานทำงาน

ต่างจาก find อย่างไร

find กรองและ project แต่คืน document ต้นฉบับ บางส่วนเสมอ aggregation ทำเช่นนั้นได้ด้วย ($match และ $project) และยังสร้าง document ใหม่ทั้งหมดได้: ยอดนับ ค่าเฉลี่ย เรคคอร์ดที่ join แล้ว array ที่ถูกคลี่

• find อ่าน document ออกมา; aggregation ปรับรูปสาย ของมัน
• find คืน document ที่มีอยู่แล้ว; aggregation สังเคราะห์ ใหม่ได้
• pipeline มีลำดับและประกอบได้ - ต่อ stage เหมือนเลโก้

ความต่างวัดได้ การหาจำนวนเล่มต่อ title นิยายด้วย find แล้วบวกในเชลล์ย้าย 40 000 document ไปยัง client; pipeline ย้ายแค่ 100 คำตอบ:

mongosh/agg-client-vs-server.js

const bytes = (docs) => docs.reduce((n, d) => n + bsonsize(d), 0);
let t = Date.now();
const all = db.sales.find({ genre: "fiction" }).toArray();              // pull every fiction order to the client ...
const totals = {};
for (const d of all) totals[d.title] = (totals[d.title] ?? 0) + d.quantity;   // ... and sum there
const clientMs = Date.now() - t;
t = Date.now();
const rows = db.sales.aggregate([{ $match: { genre: "fiction" } }, { $group: { _id: "$title", copies: { $sum: "$quantity" } } }]).toArray();
const serverMs = Date.now() - t;
console.log({ clientDocsTransferred: all.length, clientMB: +(bytes(all) / 1048576).toFixed(1), clientMs });
console.log({ serverDocsTransferred: rows.length, serverKB: +(bytes(rows) / 1024).toFixed(1), serverMs });
assert.strictEqual(rows.length, Object.keys(totals).length);
assert.ok(rows.every((r) => totals[r._id] === r.copies));

{ clientDocsTransferred: 40000, clientMB: 3.7, clientMs: 159 }
{ serverDocsTransferred: 100, serverKB: 3, serverMs: 52 }

ยอดรวมเท่ากัน (fence ยืนยัน) ไบต์บนสายน้อยกว่าราว 1 000 เท่า บนเครือข่ายจริงช่องว่างกว้างกว่ามิลลิวินาทีในเครื่องเหล่านี้มาก

pipeline ใช้ index ของคุณ

pipeline ที่เริ่มด้วย $match ถูกวาง plan เหมือน find บน filter นั้น จึงใช้ index ได้ explain แสดง plan ของ stage แรก (ส่วนที่อ่าน collection) และ stage ที่เหลือ; ที่นี่ตัว $group เองรันอยู่ใน plan เดียวกัน

mongosh/agg-explain.js

const p = [{ $match: { price: { $gte: 24 } } }, { $group: { _id: "$genre", n: { $sum: 1 } } }];
console.log("with price_1  :", aplan(p));
db.sales.dropIndex("price_1");
console.log("without index :", aplan(p));
db.sales.createIndex({ price: 1 });

with price_1  : GROUP > FETCH > IXSCAN(price_1)  keys=13422 docs=13422
without index : GROUP > COLLSCAN  keys=0 docs=200000

โมดูลนี้ครอบคลุมอะไรบ้าง

1. พื้นฐาน pipeline - $match, $project, $set ทำไมลำดับสำคัญ และ optimizer แก้ให้คุณอะไร
2. Group และ accumulator - $group, $sum, $avg, $push, $first และเพดานหน่วยความจำ
3. Lookup join - $lookup และกลยุทธ์ join ที่ server เลือก
4. Unwind array - $unwind และกรณีขอบ
5. Facet และ bucket - $facet, $bucket, $bucketAuto และขีดจำกัด

เคล็ดลับและข้อควรระวัง

• แต่ละ stage เห็นเฉพาะ output ของ stage ก่อนหน้า: field ที่ทิ้งไปเร็วจะหายสำหรับ stage หลัง
• เริ่มด้วย $match เพื่อให้ index ช่วยได้ (13 422 key แทน 200 000 document ข้างบน)
• find ธรรมดาอ่านง่ายกว่าเมื่อคุณแค่กรอง ใช้ pipeline เมื่อต้องสรุปหรือสร้าง document ใหม่

🍃 แบบจำลองความคิด

aggregation pipeline คือ array ของ stage document เข้ามา แต่ละ stage แปลงสายข้อมูลแล้วส่งต่อ และ output ของ stage สุดท้ายคือผลลัพธ์ของคุณ คิดว่าเป็นสายพาน ไม่ใช่ query เดียว

อ่านโค้ดที่ AI สร้าง

เมื่อถูกขอ “title ที่ขายดีที่สุด” AI มักเก็บทุกอย่างไว้ในโค้ดแอป:

🐛 หา bug ในโค้ดนี้

// AI comment: "load the fiction orders, then total copies per title in JavaScript"
const orders = await db.collection("sales").find({ genre: "fiction" }).toArray();
const totals = {};
for (const o of orders) totals[o.title] = (totals[o.title] ?? 0) + o.quantity;
return Object.entries(totals).sort((a, b) => b[1] - a[1]).slice(0, 3);

เฉลย

มันถูกต้อง และนั่นคือเหตุผลที่หลุดรีวิว มันลาก 40 000 document (3.7 MB, 159 ms ในการรันข้างบน) ผ่าน driver และ heap ของแอปเพื่อเก็บตัวเลขสามตัว และต้นทุนโตตาม collection ไม่ใช่ตามคำตอบ pipeline คำนวณยอดรวมเดียวกันในตัว server และคืน 100 แถว (3 KB, 52 ms) - fence ในหัวข้อก่อนหน้ายืนยันว่าทั้งสองตรงกัน ใช้ $group + $sort + $limit และให้ฐานข้อมูลพับข้อมูล

Mental Model

มอง find เป็นการเลือกแถว และ aggregate เป็นการเขียนโปรแกรม data-flow เล็ก ๆ: แต่ละ stage คือฟังก์ชันจากสาย document ไปยังสาย document โปรแกรมที่ถูกทำให้สายเล็กลงเร็ว ($match) แล้วพับ ($group) และค่อยปรับรูปให้ client

ข้อแลกเปลี่ยน

ทางเลือก	ข้อดี	ต้นทุน
aggregation pipeline	สรุปฝั่ง server: 100 แถวแทน 40 000 document	อ่านยากกว่า query ธรรมดา; debug ทีละ stage
find + application code	ง่าย; ใช้ logic ใดก็ได้	ต้องย้ายทุก document ก่อน: 3.7 MB, 159 ms เทียบ 3 KB, 52 ms
pipeline หลาย stage	ประกอบทีละขั้น	ลำดับกระทบทั้งความเร็วและความถูกต้อง

Common Misconceptions

• “aggregation แทน find ได้เสมอ” - สำหรับการกรองธรรมดา find ชัดกว่าและเร็วเท่ากัน
• “ทุก stage เห็น collection ต้นฉบับ” - แต่ละอันเห็นเฉพาะ output ของ stage ก่อนหน้า
• “aggregation ไม่สนใจ index” - $match นำหน้าใช้ price_1 (13 422 key); ไม่มี index สแกน 200 000 document
• “บวกในแอปก็ได้ ผลเหมือนกัน” - ผลเหมือนกัน แต่ไบต์มากกว่า 1 000 เท่า

ตัวอย่างจากของจริง

Keller Williams (real estate) - ทีมข้อมูลใช้ Aggregation Pipeline Builder ใน MongoDB Compass สร้าง pipeline สรุปผลบนเรคคอร์ดรายการอสังหาฯ กว่า 70 ล้านรายการบน MongoDB Atlas แทนการย้ายข้อมูลดิบทั้งหมดเข้าเลเยอร์แอปเพื่อประมวลผล (mongodb.com/customers/keller-williams).

แท็บ Python/Go/Rust สะท้อนพฤติกรรมที่พิสูจน์แล้วของ shell/Node; แต่ละแท็บถูกรันหนึ่งครั้งในโปรแกรม scratch กับ server เดียวกัน (pipeline match-group-sort เดียวกันคืน title ชุดเดียวกัน) แต่ไม่ถูกรันโดยตัวตรวจอัตโนมัติของคอร์ส

stage หนึ่งของ aggregation รับอะไรเป็น input

collection ต้นฉบับทุกครั้ง document ที่ stage ก่อนหน้าผลิตออกมา ตัวอย่างสุ่มของ collection เฉพาะ document ที่ตรงกับ stage สุดท้าย

ความสามารถใดที่ aggregation มีแต่ find ธรรมดาไม่มี

กรอง document ตามค่าของ field คืนเฉพาะบาง field สรุป document จำนวนมากเป็น document ใหม่ที่คำนวณได้ เรียงผลลัพธ์

เมื่อเรียก aggregate เขียน pipeline อย่างไร

เป็น document เดียวที่มี operator เดียว เป็น array ของ stage object ที่มีลำดับ เป็น string ชื่อ stage คั่นด้วยจุลภาค เป็น filter object ซ้อนกัน

ทำไมจึงคำนวณยอดรวมต่อ title ใน pipeline แทนหลัง find

find คืนตัวเลขไม่ได้ server คืน 100 แถวเล็กแทน 40 000 document (3 KB แทน 3.7 MB ที่นี่) aggregation เร็วกว่า find เสมอ โค้ดฝั่ง client วนผลลัพธ์ไม่ได้

pipeline เริ่มด้วย $match บน field ที่มี index explain แสดงอะไร

COLLSCAN เพราะ aggregation ไม่สนใจ index IXSCAN ที่ป้อนส่วนที่เหลือของ pipeline (13 422 key แทนการสแกน 200 000 document) error covered query

Submit

Mark as complete