aggregation pipeline

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

pipeline เป็นแนวคิดที่ทรงพลังที่สุดใน MongoDB รองจากตัว document เอง ลองนึกภาพสายพานลำเลียง: document เข้ามาทางปลายด้านหนึ่ง ไหลผ่านชุดของ stage แล้ว stream ที่แปลงเสร็จแล้วออกมาทางปลายอีกด้าน แต่ละ stage รับ document ที่ส่งมาให้ ทำงานอย่างเดียวที่นิยามไว้ชัดเจน แล้วส่งผลลัพธ์ต่อไปยัง stage ถัดไป คุณเป็นคนเลือกลำดับเอง และลำดับก็สำคัญ — stage หนึ่งเห็นเฉพาะสิ่งที่ stage ก่อนหน้าผลิตออกมาเท่านั้น

ตลอดทั้งโมดูลนี้เราจะทำงานกับชุดข้อมูลเล็ก ๆ ที่สมมติขึ้น: collection sales เก็บออเดอร์หนังสือของร้านในจินตนาการ ข้อมูลยิ่งเล็ก รูปร่าง ของการแปลงแต่ละครั้งยิ่งเห็นชัด

pipeline คือลิสต์ของ stage

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

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

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

array ของ stage ชุดเดียวกันนี้คือสิ่งที่คุณส่งเข้าไป ไม่ว่าจะใช้ภาษาไหน สังเกตว่าเมธอดที่เรียกคือ aggregate และ argument คือ pipeline:

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

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

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

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

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

จากออเดอร์ fiction ไม่กี่รายการ pipeline คืนหนึ่ง document ต่อหนึ่งชื่อเรื่อง พร้อมผลรวมจำนวนเล่ม โดยเรียงชื่อเรื่องที่ขายดีที่สุดไว้บนสุด:

[
  { "_id": "Dune", "copies": 7 },
  { "_id": "Neuromancer", "copies": 4 },
  { "_id": "Foundation", "copies": 2 }
]

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

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

find filter และ project ได้ แต่คืนได้แค่สับเซตของ document ต้นฉบับ เสมอ ส่วน aggregation ทำแบบนั้นได้เหมือนกัน และยังสร้าง document ใหม่ทั้งก้อนขึ้นมาได้ด้วย: จำนวนนับ ค่าเฉลี่ย ระเบียนที่ join แล้ว หรือ array ที่แบนราบแล้ว อะไรที่ find ทำได้ aggregation ก็ทำได้ด้วย $match และ $project แต่ทางกลับกันไม่จริง

find อ่าน document ออกมา ส่วน aggregation **จัดรูป stream ** ของ document ใหม่
find คืน document ที่มีอยู่แล้ว ส่วน aggregation สังเคราะห์ document ใหม่ขึ้นมาได้
pipeline เรียงลำดับและประกอบกันได้ — คุณเรียงซ้อน stage เหมือนต่อ Lego และเพิ่มได้เรื่อย ๆ

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

บทเรียนข้างหน้าจะพา stage ที่คุณจะหยิบมาใช้บ่อยที่สุดมาแนะนำทีละแนวคิด:

Pipeline basics — $match filter ตั้งแต่เนิ่น ๆ $project จัดรูปใหม่ และทำไมลำดับของ stage ถึงเปลี่ยนทั้งความถูกต้องและความเร็ว
Group and accumulators — $group คู่กับ $sum, $avg, $min, $max, $push และพวกพ้อง บวกกับยอดรวมทั้งหมด
Lookup joins — $lookup ดึง document ที่เกี่ยวข้องจาก collection อื่น
Unwind arrays — $unwind แตก document เดียวที่มี array ออกเป็นหลาย document
Facets and buckets — $facet ตอบหลายคำถามในการประมวลผลครั้งเดียว และ $bucket จัดกลุ่มแบบฮิสโทแกรม

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

ตัวเลือก	Benefit	Cost
aggregation pipeline	คำนวณและสรุปข้อมูลฝั่ง server ลดข้อมูลที่ต้องส่งผ่านเครือข่าย	pipeline อ่านยากกว่า query ธรรมดา และ debug ทีละ stage ต้องใช้เวลา
find + ประมวลผลใน application code	เขียนง่าย เข้าใจง่าย เหมาะกับ logic ที่ซับซ้อนซึ่งยากจะแสดงเป็น stage	ต้องดึง document ทั้งหมดผ่านเครือข่ายก่อนคำนวณ ทำให้ช้าและกิน memory เมื่อข้อมูลใหญ่
pipeline ที่มีหลาย stage	ประกอบ stage ทีละอันได้ยืดหยุ่น ต่อยอดง่าย	ยิ่งมี stage เยอะ ยิ่งต้องระวังลำดับที่ส่งผลต่อ performance และความถูกต้อง

ข้อผิดพลาดที่พบบ่อย

คิดว่า aggregation pipeline แทน find ได้เสมอ — ถ้าคุณแค่อยากได้ document ที่ตรง filter โดยไม่ต้องแปลงรูป find ธรรมดาอ่านง่ายกว่าและมักเร็วพอ ๆ กัน ใช้ pipeline ตอนที่ต้องสรุปหรือสร้าง document ใหม่จริง ๆ
มองว่า stage แต่ละตัวเห็น collection ต้นฉบับ — ทุก stage เห็นเฉพาะผลลัพธ์ของ stage ก่อนหน้า ไม่ใช่ collection ทั้งก้อน ถ้า stage ก่อนหน้าทิ้ง field ไปแล้ว stage ถัดมาก็อ้าง field นั้นไม่ได้
ลืมว่า pipeline คือ array ที่ลำดับสำคัญ — สลับตำแหน่ง stage สองตัวอาจได้ผลลัพธ์คนละแบบหรือช้าลงมาก การจัดลำดับ stage ไม่ใช่แค่เรื่องความสวยงามของโค้ด

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

Analytics dashboard ทั่วไป — หน้า dashboard ที่ต้องแสดงยอดขายรวม ยอดขายแยกตามหมวด และ top product ในคำขอเดียว มักใช้ aggregation pipeline ($group, $facet) แทนการดึง document ทั้งหมดมาคำนวณใน backend เพราะลด round-trip และภาระของ application server

Keller Williams (real estate) — ระบบรายงานอสังหาริมทรัพย์ที่ต้องสรุปยอดขาย นายหน้า และพื้นที่ ใช้ aggregation pipeline เพื่อคำนวณสรุปจากข้อมูลธุรกรรมจำนวนมากฝั่ง server แทนที่จะย้ายข้อมูลดิบทั้งหมดมาประมวลผลที่ application layer