Lab — thiết kế batch pipeline top-sản-phẩm nhiều bước

Nhóm data của một sàn thương mại điện tử vừa phát hiện báo cáo "top sản phẩm tháng" sai vì pipeline cũ join trực tiếp trong database OLTP — timeout sau 40 phút với 40 triệu dòng, và chạy lại lần hai cho kết quả khác vì bảng nguồn đã bị sửa. Bài lab này đặt bạn vào vị trí kỹ sư thiết kế lại toàn bộ từ đầu — theo đúng kỷ luật batch.

Đây là bài tổng hợp toàn module: bạn áp dụng đồng thời nguyên tắc input bất biến (bài 01), logic shuffle và external sort của MapReduce (bài 02), lý luận chọn join strategy (bài 03), và ranh giới materialization cùng fault tolerance bằng tính lại (bài 04). fadeLevel: independent có nghĩa bạn phải tự thiết kế trước khi xem gợi ý.

1. 🎯 Đề bài

Bạn nhận hai input bất biến:

Yêu cầu output: report/2026-07/top10_by_category/ với schema (category, rank, product_id, total_amount) — 10 sản phẩm có tổng doanh thu cao nhất trong mỗi danh mục, tính trên đơn đã thanh toán (status = "PAID").

Ba việc bạn cần tự làm:

(a) Vẽ chuỗi job — mỗi bước ghi rõ: đọc input gì, thực hiện một thao tác (map / join / aggregate / rank), ghi output ra đâu. Mỗi intermediate output là derived data bất biến cho bước tiếp theo, phải chạy lại được độc lập nếu bước sau lỗi.

(b) Chọn chiến lược join cho bước ghép orders với products và lập luận theo số liệu: so sánh kích thước dữ liệu thật với threshold của từng chiến lược (sort-merge / broadcast hash / partitioned hash), cân nhắc nguy cơ skew trên join key.

(c) Ước lượng I/O của toàn pipeline tính bằng GB — bước join sort-merge tốn bao nhiêu? Tổng pipeline khuếch đại dữ liệu bao nhiêu lần so với 10 GB input gốc?

2. 🔍 I-P-O

3. 📦 Concept mapping

Bài lab này là bằng chứng trực tiếp cho hai learning outcome cuối của module:

• LO3 — Compare join strategies (bài 03 — join trong batch): yêu cầu (b) buộc bạn chọn và lập luận sort-merge vs broadcast vs partitioned hash theo kích thước thật và rủi ro skew — không chỉ nhớ tên.
• LO4 — Design pipeline + FT (bài 04 — dataflow và fault tolerance): yêu cầu (a) và (c) buộc bạn vẽ chuỗi bước với ranh giới materialization rõ ràng, ước lượng I/O từ số pass external sort, và lập luận tại sao mỗi intermediate output là điểm phục hồi.

Hai outcome được gọi lại để tổng hợp:

• LO1 (bài 01 — tư duy batch): mỗi intermediate output phải là derived data — bước sau đọc read-only, bước trước không bị sửa.
• LO2 (bài 02 — MapReduce): sort-merge join dùng external sort phân tán; shuffle là bước sort theo join key.

4. ▶️ Starter — tự thiết kế trước

Thay mọi [TODO] bằng câu trả lời của bạn. Trong pseudocode, comment sau # là nơi ghi lý luận.

PIPELINE monthly_top10(month):

  # --- Buoc 1: loc va chieu (map-only, khong shuffle) ---
  step1 <- read_only("orders/" + month + "/")  # BAT BIEN — chi doc
             .filter(o -> [TODO: dieu kien loc don hang nao?])
             .project(o -> [TODO: chi giu field nao? vi sao?])
  write("work/step1/", step1)
  # Uoc tinh kich thuoc output: [TODO] GB (vi sao?

  # --- Buoc 2: join de lay category ---
  products <- read_only("products/snapshot/")   # BAT BIEN
  # Chien luoc join: [TODO: sort-merge / broadcast / partitioned-hash]
  # Ly do: step1 = [TODO] GB, products = 500 MB
  #         Broadcast threshold mac dinh: 10 MB; hard limit: 8 GB
  #         -> [TODO: tai sao chon hoac khong chon broadcast?]
  #         Skew tren join key product_id: [TODO: co nguy co khong? tai sao?]
  joined <- join(step1, products, key = "product_id", strategy = [TODO])
  write("work/step2/", joined)

  # --- Buoc 3: tong hop ---
  aggregated <- read_only("work/step2/")
                  .group_by([TODO: group key la gi?])
                  .aggregate([TODO: ham tong hop nao?])
  write("work/step3/", aggregated)
  # Uoc tinh so ban ghi output: [TODO] (vi sao?)

  # --- Buoc 4: xep hang moi danh muc ---
  ranked <- read_only("work/step3/")
              .[TODO: group theo gi, sort theo gi, lay bao nhieu?]
  write("report/" + month + "/top10_by_category/", ranked)

  # Neu buoc 2 chet giua chung:
  # -> viec can lam: [TODO]
  # -> co phai chay lai tu buoc 1 khong? [TODO] Vi sao?

# ----------------------------------------------------------------
# Uoc luong I/O toan pipeline:
#
#  Buoc 1 (filter+project): doc [TODO] GB + ghi [TODO] GB = [TODO] GB
#
#  Buoc 2 (sort-merge join):
#    Sort step1 ([TODO] GB, gia su RAM per task = 512 MB):
#      So initial sorted runs = [TODO] GB / 512 MB = [TODO] runs
#      So merge pass can thiet: [TODO]
#      => [TODO] pass x [TODO] GB x 2 (doc+ghi) = [TODO] GB
#    Sort products (500 MB, vua RAM):
#      => [TODO] GB
#    Merge join pass (doc ca 2 da sort, ghi ket qua):
#      => [TODO] GB
#    Buoc 2 tong: [TODO] GB
#
#  Buoc 3 (shuffle aggregate): doc [TODO] GB + ghi [TODO] GB = [TODO] GB
#  Buoc 4 (rank): nho, bo qua
#
#  TONG PIPELINE: ~[TODO] GB I/O cho 10 GB input goc
#  Boi so khuech dai: ~[TODO]x

5. 💡 Gợi ý

5.1 Chuỗi job — tại sao nên chiếu fields sớm?

Trước khi join, hỏi: output của join cần gì từ orders? Sau khi ghép với products và tổng hợp, kết quả cuối chỉ cần product_id và amount từ bảng orders — không cần order_id, qty, status, ts. Nếu chiếu ngay ở bước 1 (trước join), kích thước data orders đi vào join giảm đáng kể — ít I/O ở bước đắt nhất.

Từ (product_id, category, amount) sau join tới (category, rank, product_id, total_amount) cần hai bước nữa: gom và cộng dồn theo (category, product_id), rồi xếp hạng trong từng danh mục. Hai bước này cần shuffle khác nhau (hoặc gộp nếu engine hỗ trợ) nên tách cho rõ ranh giới materialization.

5.2 Join strategy — ba câu hỏi phải trả lời

Câu 1 — broadcast có an toàn không? Broadcast hash join copy bảng nhỏ lên RAM của tất cả worker đồng thời. Threshold mặc định của Spark là 10 MB; hard limit là 8 GB. Công thức: với C worker, tổng RAM cho broadcast = C × (kích thước bảng nhỏ). Tự tính: products 500 MB trên ~50 worker tốn bao nhiêu RAM, so với RAM mỗi node (8–16 GB) — có an toàn để broadcast không? (Chốt ở §6.)

Câu 2 — có skew trên join key không? Join key là product_id. Skew xảy ra khi một số product_id có hàng triệu dòng trong orders (hot-seller). Nếu có skew: sort-merge join vẫn an toàn (straggler chậm nhưng không crash); partitioned hash join có thể OOM nếu một partition quá lớn.

Câu 3 — kích thước hai bảng sau khi lọc? Bước 1 đã chiếu orders xuống còn (product_id, amount). Tự ước lượng: còn khoảng bao nhiêu GB? Products = 500 MB. So cả hai với broadcast threshold — bảng nào (nếu có) đủ nhỏ để broadcast an toàn? Nếu cả hai vượt ngưỡng, chiến lược nào vừa chịu được skew của product_id vừa không OOM? (Chốt đáp án ở phần Lời giải §6.)

5.3 Ước lượng I/O — áp công thức external sort

Sort-merge join cần sort cả hai input theo join key trước khi merge. Chi phí chính là external sort của bảng lớn hơn.

Công thức external sort (bài thuật toán + bài 02): với dữ liệu kích thước N, RAM buffer M mỗi task, fan-in k:

• Bước tạo sorted runs: đọc N, ghi N → 2N I/O
• Mỗi merge pass: đọc N, ghi N → 2N I/O; số pass = ⌈log_k(N/M)⌉

Giờ tự áp số của bạn (đừng xem Lời giải vội): với step1 ≈ ? GB, RAM/task giả sử 512 MB, k ≥ 8 → số run ban đầu = ?, số merge pass = ⌈log_k(N/M)⌉ = ?, tổng I/O external sort step1 = ? Rồi cộng chi phí sort products (vừa RAM) và merge join pass. Đối chiếu kết quả với bảng I/O ở §6.

Cộng chi phí cả bốn bước lại, bạn sẽ ra tổng I/O của pipeline và hệ số khuếch đại so với 10 GB input gốc — đối chiếu với bảng tổng hợp ở §6.

6. ✅ Lời giải tham khảo

6.1 Chuỗi job

flowchart TD
  O["orders/2026-07/ -- 10 GB (BAT BIEN)"]
  P["products/snapshot/ -- 500 MB (BAT BIEN)"]

  S1["Step 1: filter PAID + project (product_id, amount) -- map only"]
  W1["work/step1/ -- ~4 GB (DERIVED, bat bien cho step 2)"]

  S2["Step 2: sort-merge join -- key: product_id"]
  W2["work/step2/ -- ~4 GB (category, product_id, amount)"]

  S3["Step 3: group by (category, product_id) -- SUM(amount)"]
  W3["work/step3/ -- ~100 MB (category, product_id, total_amount)"]

  S4["Step 4: rank per category -- top-10 by total_amount DESC"]
  R["report/2026-07/top10_by_category/ -- vai MB (PHAI SINH cuoi)"]

  O --> S1 --> W1 --> S2
  P --> S2
  S2 --> W2 --> S3 --> W3 --> S4 --> R

6.2 Pseudocode đầy đủ

PIPELINE monthly_top10(month):

  # --- Step 1: filter + project (map-only, no shuffle) ---
  # Chieu som: bo order_id, qty, status, ts truoc khi join
  # -> giam orders tu 10 GB xuong ~4 GB (bao gom loc ~70% don PAID)
  step1 <- read_only("orders/" + month + "/")      # BAT BIEN
             .filter(o -> o.status = "PAID")        # chi don thanh toan
             .project(o -> (o.product_id, o.amount)) # chi giu 2 field can thiet
  write("work/step1/", step1)                       # PHAI SINH ~4 GB

  # --- Step 2: sort-merge join (shuffle on product_id) ---
  # Ly do chon sort-merge:
  #   step1 = ~4 GB >> broadcast threshold (10 MB Spark default, hard limit 8 GB)
  #   products = 500 MB >> nguong an toan broadcast (50 worker x 500 MB = 25 GB RAM)
  #   Khong co thong tin skew cuc doan tren product_id -> sort-merge an toan
  products <- read_only("products/snapshot/")        # BAT BIEN
  joined <- sort_merge_join(step1, products, key = "product_id")
  write("work/step2/", joined)                       # (product_id, category, amount) ~4 GB

  # --- Step 3: aggregate (shuffle on (category, product_id)) ---
  aggregated <- read_only("work/step2/")
                  .group_by(r -> (r.category, r.product_id))
                  .aggregate(SUM(r.amount) as total_amount)
  write("work/step3/", aggregated)
  # Output: (category, product_id, total_amount)
  # So ban ghi <= 5M (so san pham co trong orders); thuc te ~2-3M, ~100 MB

  # --- Step 4: rank per category (shuffle on category) ---
  ranked <- read_only("work/step3/")
              .group_by(r -> r.category)
              .sort_desc(r -> r.total_amount)   # sap giam dan trong moi nhom
              .rank_and_take(n = 10)             # top-10 moi danh muc + gan rank
  write("report/" + month + "/top10_by_category/", ranked)

  # --- Fault tolerance ---
  # Step 2 chet giua chung:
  #   1. Xoa sach work/step2/ (output ghi mot phan, KHONG dung)
  #   2. Chay lai chi tu Step 2: doc work/step1/ (bat bien, van nguyen) + products
  #   3. KHONG can chay lai Step 1 vi work/step1/ da hoan thanh va bat bien
  # Nguyen tac: moi buoc chi commit output khi hoan thanh (atomic rename / _SUCCESS file)

6.3 Lập luận join strategy

Tại sao sort-merge join?

Broadcast hash join yêu cầu bảng nhỏ vừa RAM đồng thời trên tất cả worker. Với 500 MB và 50 worker: 25 GB RAM bị chiếm. Trên cluster thực tế (worker 16 GB RAM), điều này gây áp lực bộ nhớ nghiêm trọng và dễ OOM. Threshold mặc định 10 MB của Spark tồn tại chính xác vì lý do này — không phải ngẫu nhiên.

Partitioned hash join chia cả hai bảng thành N partition theo hash của join key, rồi join từng cặp partition trong RAM. Với step1 = 4 GB / 200 partition = 20 MB mỗi partition — vừa RAM — về lý thuyết khả thi. Nhưng nếu có hot-seller (một product_id chiếm nhiều triệu đơn), partition đó vượt RAM và crash. Sort-merge join không có giới hạn này: straggler chỉ chậm, không crash.

Kết luận: sort-merge join — an toàn cho cả hai bảng lớn, không nhạy cảm với skew join key, là lựa chọn mặc định của Spark cho hai bảng vượt broadcast threshold.

6.4 Ước lượng I/O

Pipeline tiêu thụ khoảng 44 GB I/O để xử lý 10 GB input — hệ số khuếch đại 4.4×. Phần đắt nhất là Step 2 (sort-merge join ≈ 25.5 GB, chiếm hơn nửa tổng I/O). So sánh: nếu products nhỏ hơn 10 MB (broadcast an toàn), Step 2 chỉ tốn ~8 GB (đọc step1 4 GB + ghi joined 4 GB; bảng nhỏ broadcast không đáng kể, không sort) — giảm khoảng 2/3 chi phí join.

7. 🎓 Mở rộng

Ba tình huống để suy nghĩ thêm sau khi đã hoàn thành lời giải tham khảo:

M1 — Hot-key skew: Sản phẩm P-0001 (áo thun cơ bản) xuất hiện trong 5 triệu trên 40 triệu đơn. Trong sort-merge join, task nhận key P-0001 xử lý 5 triệu record — gấp vài nghìn lần task trung bình → straggler kéo dài cả job. Phương án: kỹ thuật salted join thêm suffix ngẫu nhiên vào join key của bảng lớn, replicate bảng nhỏ tương ứng, rồi merge kết quả. Trade-off là gì?

M2 — Products thay đổi liên tục: Nếu products/ không phải snapshot bất biến mà được update theo giờ (thêm sản phẩm mới, đổi tên category), pipeline trên có còn idempotent không? Chạy lại vào tháng 8 với input tháng 7 cho cùng kết quả không? Bạn sẽ sửa thiết kế ở đâu?

M3 — Thêm chiều thời gian: Ban quản lý muốn báo cáo theo tuần thay vì tháng, nhưng vẫn dùng cùng log theo ngày. Bạn sẽ tổ chức input và pipeline như thế nào để tránh đọc lại toàn bộ dữ liệu tháng khi chỉ cần dữ liệu một tuần?

8. ✨ Điều bạn vừa làm được

• Áp dụng input bất biến xuyên suốt pipeline (bài 01): mỗi bước chỉ đọc output bất biến của bước trước; intermediate output là điểm phục hồi độc lập. Nếu Step 2 lỗi, bạn xoá work/step2/ và chạy lại từ Step 2 — không phải từ đầu. Đây là lý do mỗi derived data được ghi ra nơi tách biệt.

• Suy luận về join strategy bằng số liệu (bài 03): không chỉ nhớ "broadcast cho bảng nhỏ" mà tính cụ thể — 500 MB × 50 worker = 25 GB RAM, vượt ngưỡng an toàn. Kết luận sort-merge join bằng lý luận định lượng, không phải trực giác.

• Ước lượng I/O từ số pass external sort (bài 02 + bài thuật toán): áp công thức ⌈log_k(N/M)⌉ để tính số merge pass, nhân với kích thước data, tổng hợp các bước → thấy join chiếm hơn nửa tổng I/O của pipeline.

• Xác định ranh giới materialization (bài 04): hiểu tại sao phải ghi đĩa sau shuffle boundary (Step 2 — shuffle theo product_id; Step 3 — shuffle theo category), và tại sao engine DAG có thể pipeline Step 3→Step 4 trong cùng một stage mà không cần intermediate file.

Bài tiếp theo: Tổng kết module — cheat sheet & self-assessment