NoSQL 3 — MongoDB: CRUD & Aggregation Pipeline

Ở bài tổng quan NoSQL chúng ta đã phân loại các họ NoSQL, và ở bài mô hình dữ liệu MongoDB đã thiết kế document, phân biệt embed và reference. Bài này bước vào phần thao tác thực tế: đọc/ghi dữ liệu và đặc biệt là Aggregation Pipeline — thứ khiến MongoDB không chỉ là "kho JSON" mà là một công cụ phân tích thực thụ.

Người viết giả định bạn đã quen SQL. Vì vậy xuyên suốt bài, mỗi thao tác Mongo sẽ được đối chiếu với câu SQL tương đương để bạn "dịch" nhanh trong đầu. Bối cảnh dữ liệu là bài toán ngân hàng quen thuộc: tổng hợp giao dịch theo khách hàng, theo loại, theo thời gian.

Lưu ý về code trong bài: Toàn bộ ví dụ Mongo shell (mongosh) và pipeline JSON dưới đây là cú pháp MongoDB, không phải SQL chạy được trên sandbox. Chỉ có duy nhất một block được đánh dấu -- ▶ Chạy được ở cuối phần Aggregation là câu SQL PostgreSQL thật, dùng để đối chiếu. Đừng dán nhầm pipeline JSON vào ô SQL.

Giả sử ta có collection transactions với các document dạng:

// một document trong collection transactions
{
  _id: ObjectId("..."),
  account_id: 1042,
  customer: { id: 77, name: "Nguyễn Văn A", city: "Hà Nội" },
  amount: 3500000,
  kind: "transfer",          // transfer | deposit | withdraw | fee
  channel: "mobile",
  tags: ["payroll", "vip"],
  created_at: ISODate("2026-06-01T09:12:00Z")
}

1. Insert — thêm document

insertOne thêm một document, insertMany thêm nhiều trong một lệnh (mặc định ordered: gặp lỗi ở document nào thì dừng, các document sau không được ghi; đặt { ordered: false } để ghi tiếp phần còn lại).

db.transactions.insertOne({
  account_id: 1042, amount: 3500000, kind: "transfer",
  channel: "mobile", created_at: new Date()
})

db.transactions.insertMany([
  { account_id: 1042, amount: 200000, kind: "fee",     created_at: new Date() },
  { account_id: 2001, amount: 5000000, kind: "deposit", created_at: new Date() }
])

Tương đương SQL: INSERT INTO transactions (...) VALUES (...). Khác biệt cốt lõi — MongoDB không ép schema: hai document trong cùng collection có thể khác tập trường. Nếu không truyền _id, MongoDB tự sinh ObjectId (12 byte: timestamp + máy + tiến trình + bộ đếm), đảm bảo duy nhất mà không cần sequence tập trung như SERIAL.

2. Find — truy vấn với query filter

find(filter, projection) là SELECT ... WHERE. Filter là một document; mỗi cặp field: value mặc định là điều kiện bằng ($eq) và nhiều cặp được AND ngầm.

Query operator thường dùng

// giao dịch chuyển tiền >= 1 triệu ở Hà Nội
db.transactions.find({
  kind: "transfer",
  amount: { $gte: 1000000 },
  "customer.city": "Hà Nội"
})

// kind thuộc tập, HOẶC amount rất lớn
db.transactions.find({
  $or: [
    { kind: { $in: ["transfer", "withdraw"] } },
    { amount: { $gt: 50000000 } }
  ]
})

// tên khách bắt đầu bằng "Nguyễn" (không phân biệt hoa thường)
db.transactions.find({ "customer.name": { $regex: /^Nguyễn/, $options: "i" } })

Chú ý cú pháp "customer.city" — dot notation để truy vào trường lồng nhau trong sub-document, thứ SQL thuần không có. $regex mạnh nhưng nếu không neo đầu chuỗi (^) sẽ không dùng được index → quét toàn bộ; chi tiết ở bài index & hiệu năng.

Một khác biệt tinh tế mà người quen SQL hay vấp: khi filter một trường là mảng, ví dụ tags: "vip", MongoDB hiểu là "mảng tags có chứa phần tử vip" — không phải "trường tags bằng chuỗi vip". Đây là hành vi mặc định rất tiện cho dữ liệu dạng mảng; nếu cần khớp chính xác cả mảng thì dùng $eq với một mảng đầy đủ, còn khớp nhiều điều kiện trên cùng phần tử mảng thì dùng $elemMatch. Ngoài ra, khi so sánh với null, field: null khớp cả trường bằng null lẫn trường không tồn tại — muốn phân biệt phải kết hợp $exists.

Projection — chọn cột trả về

Tham số thứ hai của find là projection: 1 để lấy trường, 0 để bỏ. _id mặc định luôn trả, muốn ẩn phải nêu _id: 0. Không được trộn lẫn 1 và 0 trong cùng projection (ngoại trừ _id).

db.transactions.find(
  { kind: "transfer" },
  { _id: 0, account_id: 1, amount: 1, created_at: 1 }   // ~ SELECT account_id, amount, created_at
)

Sort, limit, skip

db.transactions.find({ account_id: 1042 })
  .sort({ created_at: -1 })   // -1 giảm dần (DESC), 1 tăng dần (ASC)
  .skip(20)                    // OFFSET 20
  .limit(10)                   // LIMIT 10

Đây chính là ORDER BY created_at DESC LIMIT 10 OFFSET 20. Cảnh báo giống SQL: skip lớn (phân trang sâu) rất tốn kém vì Mongo vẫn phải duyệt qua và bỏ N document đầu; với dữ liệu lớn nên phân trang bằng con trỏ (created_at < <giá trị cuối trang trước>) thay vì skip.

3. Update — cập nhật với update operator

Điểm bẫy lớn nhất cho người mới: updateOne(filter, update) không nhận một document thay thế ở tham số update — nó nhận update operator. Viết updateOne({...}, { amount: 0 }) sẽ báo lỗi; phải dùng $set.

// updateOne: cập nhật document ĐẦU TIÊN khớp filter
db.transactions.updateOne(
  { _id: ObjectId("...") },
  { $set: { channel: "internet-banking" }, $inc: { retry_count: 1 } }
)

// updateMany: cập nhật MỌI document khớp  ~  UPDATE ... WHERE ...
db.transactions.updateMany(
  { kind: "fee", amount: { $lt: 1000 } },
  { $set: { waived: true } }
)

// thao tác trên mảng tags
db.transactions.updateOne({ _id: ObjectId("...") },
  { $addToSet: { tags: "reviewed" },   // không thêm nếu đã có
    $pull:     { tags: "vip" } })       // xoá tag "vip"

Upsert — chèn nếu chưa có

{ upsert: true } biến update thành "cập nhật nếu tồn tại, chèn mới nếu không" — tương đương INSERT ... ON CONFLICT DO UPDATE của PostgreSQL. Cực hữu ích cho các bảng tổng hợp cập nhật dần (running total).

// tăng tổng chi tiêu ngày của tài khoản; tạo dòng nếu chưa có
db.daily_totals.updateOne(
  { account_id: 1042, day: "2026-06-30" },
  { $inc: { total: 3500000, txn_count: 1 } },
  { upsert: true }
)

replaceOne vs updateOne

replaceOne(filter, doc) thay toàn bộ document (trừ _id) bằng doc mới — không dùng operator, mọi trường cũ không có trong doc sẽ mất. Dùng khi bạn có bản ghi hoàn chỉnh mới; còn sửa vài trường thì luôn dùng updateOne + $set.

4. Delete

db.transactions.deleteOne({ _id: ObjectId("...") })              // xoá 1
db.transactions.deleteMany({ created_at: { $lt: ISODate("2020-01-01") } })  // xoá nhiều

Tương đương DELETE FROM ... WHERE .... Với ngân hàng, hiếm khi xoá cứng giao dịch (yêu cầu lưu trữ pháp lý); thường "xoá mềm" bằng $set: { deleted_at: ... } hoặc chuyển sang collection archive.

5. Aggregation Pipeline — trái tim của MongoDB

Đây là phần quan trọng nhất. Aggregation Pipeline là một mảng các stage; dữ liệu chảy qua từng stage như đường ống Unix (|), mỗi stage biến đổi luồng document rồi đẩy sang stage sau. Nó làm được mọi thứ GROUP BY của SQL làm — và nhiều hơn: bung mảng, join, chạy nhiều nhánh tổng hợp song song.

Các stage cốt lõi

Các accumulator trong $group: $sum, $avg, $min, $max, $count, $push (gom giá trị vào mảng), $addToSet (gom giá trị duy nhất), $first, $last.

Ví dụ 1: tổng số tiền theo loại giao dịch

Bài toán: tổng và số lượng giao dịch theo kind, chỉ tính năm 2026, sắp theo tổng giảm dần.

db.transactions.aggregate([
  { $match: { created_at: { $gte: ISODate("2026-01-01") } } },  // lọc SỚM
  { $group: {
      _id: "$kind",                       // khoá nhóm (GROUP BY kind)
      total:  { $sum: "$amount" },        // SUM(amount)
      count:  { $sum: 1 },                // COUNT(*)
      avgAmt: { $avg: "$amount" }         // AVG(amount)
  }},
  { $sort: { total: -1 } },
  { $project: { _id: 0, kind: "$_id", total: 1, count: 1, avgAmt: 1 } }
])

Câu SQL tương đương với pipeline trên (chạy được thật trên sandbox PostgreSQL để bạn đối chiếu kết quả):

-- ▶ Chạy được
SELECT kind, SUM(amount) AS total, COUNT(*) AS count, AVG(amount) AS avg_amt
FROM transactions
WHERE created_at >= '2026-01-01'
GROUP BY kind
ORDER BY total DESC;

Hãy để ý sự tương ứng 1-1: $match → WHERE, _id: "$kind" → GROUP BY kind, $sum/$avg → SUM/AVG, $sort → ORDER BY. Người quen SQL nắm được cấu trúc này là đã đọc hiểu 80% pipeline gặp trong thực tế.

Có vài chỗ đảo ngược tư duy so với SQL cần khắc cốt. Thứ nhất, trong $group, trường khoá nhóm bắt buộc đặt tên là _id — muốn nhóm theo nhiều trường thì _id là một document { a: "$x", b: "$y" } (tương đương GROUP BY x, y). Thứ hai, ký hiệu "$field" với dấu $ đứng trước là tham chiếu giá trị của trường đó, khác với field (chuỗi tên trường) — nhầm chỗ này là lỗi phổ biến nhất khi mới học. Thứ ba, pipeline có thứ tự và có thể lặp lại stage: bạn hoàn toàn được $group hai lần liên tiếp (như ví dụ use case cuối bài), hay $match ở nhiều chỗ khác nhau — điều SQL đơn tầng không cho phép, chỉ CTE lồng nhau mới mô phỏng được.

Ví dụ 2: $lookup — join giao dịch với thông tin tài khoản

Giả sử thông tin tài khoản nằm ở collection riêng accounts (mô hình reference). $lookup nối chúng như LEFT JOIN:

db.transactions.aggregate([
  { $match: { amount: { $gte: 10000000 } } },
  { $lookup: {
      from: "accounts",              // collection cần join
      localField: "account_id",      // trường bên transactions
      foreignField: "id",            // trường bên accounts
      as: "acct"                     // kết quả là MẢNG các account khớp
  }},
  { $unwind: "$acct" },              // bung mảng 1-phần-tử thành object phẳng
  { $project: {
      _id: 0, account_id: 1, amount: 1,
      currency: "$acct.currency", balance: "$acct.balance"
  }}
])

Điểm khác biệt then chốt: $lookup luôn trả kết quả vào một mảng (as: "acct") vì về lý thuyết một khoá có thể khớp nhiều document bên from. Vì vậy gần như luôn có $unwind ngay sau để làm phẳng — trừ khi bạn cố tình muốn giữ mảng.

Ví dụ 3: $unwind — bung mảng tags rồi đếm

$unwind biến một document có mảng N phần tử thành N document. Rất hợp để "đếm theo tag":

db.transactions.aggregate([
  { $unwind: "$tags" },                          // mỗi tag một dòng
  { $group: { _id: "$tags", n: { $sum: 1 } } },  // đếm theo tag
  { $sort: { n: -1 } }
])

Ví dụ 4: $facet — nhiều báo cáo trong một lần quét

$facet chạy nhiều pipeline con song song trên cùng luồng đầu vào, trả về một document chứa mọi kết quả. Lý tưởng cho dashboard cần nhiều lát cắt mà chỉ quét dữ liệu một lần:

db.transactions.aggregate([
  { $match: { created_at: { $gte: ISODate("2026-06-01") } } },
  { $facet: {
      byKind:    [ { $group: { _id: "$kind",    total: { $sum: "$amount" } } } ],
      byChannel: [ { $group: { _id: "$channel", total: { $sum: "$amount" } } } ],
      top5:      [ { $sort: { amount: -1 } }, { $limit: 5 },
                   { $project: { _id: 0, account_id: 1, amount: 1 } } ]
  }}
])

Luồng một pipeline nhiều stage

Mỗi mũi tên là luồng document; số lượng document có thể tăng ($unwind) hoặc giảm mạnh ($match, $group) qua từng chặng. Đặt các stage thu hẹp dữ liệu càng sớm càng tốt.

Transaction đa document

Trong MongoDB, một thao tác trên một document luôn nguyên tử (atomic) — kể cả khi cập nhật nhiều trường lồng nhau. Đây là lý do mô hình embed (bài 2) giảm nhu cầu transaction. Nhưng khi cần cập nhật nhiều document/nhiều collection cùng nguyên tử, MongoDB hỗ trợ multi-document transaction có ACID từ phiên bản 4.0 (replica set) và 4.2 (sharded cluster).

const session = db.getMongo().startSession();
session.startTransaction();
try {
  const acc = session.getDatabase("bank").accounts;
  acc.updateOne({ id: 1042 }, { $inc: { balance: -1000000 } });  // trừ tài khoản A
  acc.updateOne({ id: 2001 }, { $inc: { balance:  1000000 } });  // cộng tài khoản B
  session.commitTransaction();   // cả hai cùng thành công
} catch (e) {
  session.abortTransaction();    // hoặc cùng rollback
} finally {
  session.endSession();
}

Khi nào cần transaction: chuyển khoản (trừ chỗ này cộng chỗ kia), ghi bút toán kép, cập nhật đơn hàng + tồn kho. Khi nào tránh: vì transaction đa document tốn kém (giữ khoá, tăng độ trễ, giới hạn thời gian mặc định 60 giây), hãy ưu tiên thiết kế để nghiệp vụ gói gọn trong một document khi có thể. Đừng bê nguyên tư duy "mọi thứ đều bọc BEGIN...COMMIT" của RDBMS sang. So sánh sâu về ACID nằm ở bài giao dịch & ACID trong SQL.

Cạm bẫy hiệu năng cần nhớ

• Đặt $match sớm nhất có thể. Stage đầu tiên nên là $match để cắt dữ liệu ngay; đặt nó sau $group/$lookup khiến Mongo xử lý cả đống document rồi mới vứt đi. $match ở đầu còn có thể dùng index; sau $group thì không.
• $lookup đắt. Nó thực chất là vòng lặp tra cứu; nếu foreignField không có index thì mỗi document bên trái quét toàn bộ collection phải → chậm theo cấp số nhân. Luôn index trường được join, và $match để giảm số document trước khi $lookup.
• $unwind làm phình dữ liệu. Một document mảng 100 phần tử thành 100 document; đặt $unwind sau khi đã lọc bớt.
• Giới hạn bộ nhớ 100MB mỗi stage. $group/$sort vượt ngưỡng sẽ lỗi trừ khi bật allowDiskUse: true (ghi tạm ra đĩa, chậm hơn). Tốt hơn là $sort dựa trên index và giảm dữ liệu trước.
• $project sớm để bỏ trường nặng (blob, mảng lớn) không dùng, giảm dữ liệu phải khiêng qua pipeline.

Use case thực tế

Bài toán: Phòng Phân tích khách hàng của NCB cần báo cáo hàng ngày: với mỗi khách VIP (có tag vip), tính tổng giá trị giao dịch và phân bố theo loại trong 30 ngày qua, phục vụ chấm điểm quan hệ (relationship score). Dữ liệu giao dịch nằm ở collection transactions (~40 triệu document/tháng), thông tin khách ở customers.

Pipeline triển khai:

db.transactions.aggregate([
  // 1. Lọc SỚM: 30 ngày gần nhất + chỉ giao dịch có tag vip
  { $match: {
      created_at: { $gte: ISODate("2026-06-01") },
      tags: "vip"
  }},
  // 2. Gom theo khách + loại, tính tổng & số lượng
  { $group: {
      _id: { cust: "$customer.id", kind: "$kind" },
      total: { $sum: "$amount" },
      cnt:   { $sum: 1 }
  }},
  // 3. Gom tiếp về mức khách: gộp các loại vào một mảng breakdown
  { $group: {
      _id: "$_id.cust",
      grandTotal: { $sum: "$total" },
      breakdown:  { $push: { kind: "$_id.kind", total: "$total", cnt: "$cnt" } }
  }},
  // 4. Nối tên khách để báo cáo dễ đọc
  { $lookup: { from: "customers", localField: "_id",
               foreignField: "id", as: "c" } },
  { $unwind: "$c" },
  { $project: { _id: 0, customer: "$c.name",
                grandTotal: 1, breakdown: 1 } },
  { $sort: { grandTotal: -1 } },
  { $limit: 100 }
], { allowDiskUse: true })

Kết quả & quyết định hiệu năng:

• Trên 40 triệu document, $match đầu tiên (có index trên created_at và tags) cắt xuống còn ~1,2 triệu document VIP trong 30 ngày — chỉ 3% dữ liệu đi vào các stage sau.
• $lookup sang customers chỉ chạy trên ~800 khách VIP đã gom (sau hai $group), không phải trên 1,2 triệu giao dịch → tránh được thảm hoạ join. Đây là minh hoạ trực tiếp nguyên tắc "gom trước, join sau".
• breakdown dùng $push cho ra mảng phân bố theo loại ngay trong một document kết quả — thứ mà SQL phải làm bằng nhiều dòng hoặc hàm cửa sổ rồi pivot.
• Thời gian chạy thực tế: từ vài phút (khi $match đặt cuối, phiên bản sai) rút xuống dưới 10 giây sau khi đưa $match lên đầu và thêm index. Đây là bài học kinh điển về thứ tự stage.

Ghi nhớ

• CRUD map thẳng sang SQL: find=SELECT WHERE, insertMany=INSERT, updateMany+$set=UPDATE, deleteMany=DELETE; projection = danh sách cột; sort/limit/skip = ORDER BY/LIMIT/OFFSET.
• Update phải dùng operator ($set, $inc, $push, $addToSet, $pull). updateOne({...}, {field: v}) là sai — sẽ báo lỗi. replaceOne mới thay toàn bộ document.
• upsert: true = INSERT ... ON CONFLICT DO UPDATE, lý tưởng cho bảng tổng hợp cập nhật dần bằng $inc.
• Aggregation Pipeline là chuỗi stage, mạnh hơn GROUP BY: $match(WHERE) → $group(GROUP BY + $sum/$avg/$push) → $lookup(JOIN) → $unwind(bung mảng) → $sort/$project. $facet chạy nhiều nhánh song song.
• Hiệu năng: đặt $match sớm nhất (dùng được index), gom trước rồi mới $lookup (và luôn index foreignField), cẩn thận $unwind làm phình dữ liệu, để ý giới hạn 100MB/stage.
• Transaction đa document có ACID từ 4.0 — dùng cho chuyển khoản/bút toán kép, nhưng tốn kém; ưu tiên thiết kế gói nghiệp vụ trong một document (thao tác một document luôn nguyên tử).

Bài tiếp theo đi sâu vào index & hiệu năng MongoDB — thứ quyết định pipeline chạy trong giây hay trong phút. Sau đó series chuyển sang Redis: khái niệm cơ bản, pattern, vận hành và tổng hợp use case ngân hàng.