Join trong batch — sort-merge, broadcast, partitioned

TL;DR: Join trong batch quy về ba chiến lược. Sort-merge join sort cả hai bảng theo join key rồi chạy k-way merge — xử lý được bảng terabyte bất kể skew, chi phí là hai lần sort. Broadcast hash join copy bảng nhỏ vào RAM mọi executor và probe hash table — tránh shuffle bảng lớn, nhưng chỉ an toàn khi bảng nhỏ thực sự vừa RAM (Spark mặc định 10 MB, hard limit 8 GB). Partitioned hash join chia cả hai bảng theo cùng hàm hash — partition nào join độc lập, nhưng hot key skew dồn hàng triệu record về một reducer. Pitfall kinh điển: broadcast bảng "nhỏ" mà thực chất GB-scale, và hash(hot_key) dồn gấp hàng trăm tới hàng nghìn lần record so với mức trung bình về một nút.

Hệ thống đề xuất của một sàn thương mại điện tử cần join bảng events (2 TB click/view log) với bảng products (8 GB metadata: tên, giá, danh mục). Một sản phẩm hot có thể xuất hiện trong 50 triệu dòng log — trong khi sản phẩm trung bình chỉ vài trăm. Câu hỏi đặt ra: chọn chiến lược join nào? Broadcast bảng products 8 GB lên RAM của 50 executor sẽ ngốn 400 GB RAM — một tính toán có thể làm cả cluster sập. Sort-merge thì sao? Partitioned hash thì coi chừng gì?

Bài này giải thích ba chiến lược join trong batch, cơ chế bên dưới từng cái, và nguyên tắc lựa chọn theo kích thước dữ liệu và skew — vì chọn sai không chỉ chậm mà có thể làm job không bao giờ về đích.

1. Analogy — thư viện tra chéo

Thư viện có hai kệ sách: kệ A "Đơn hàng" (2 triệu cuốn) và kệ B "Sản phẩm" (500.000 cuốn). Bạn cần tra chéo: với mỗi đơn hàng, lấy thông tin sản phẩm tương ứng.

Ba cách tiếp cận khác nhau:

Sort-merge: Sắp cả hai kệ theo mã sản phẩm (một lần). Sau đó hai thủ thư đi song song từ đầu — gặp mã trùng thì ghi lại. O(n log n) lúc sort, rồi O(n+m) lúc merge.

Broadcast hash: Photo toàn bộ kệ B (nhỏ), dán một bản lên mỗi bàn làm việc. Thủ thư tra kệ A rồi tra hash table ngay tại bàn, không cần di chuyển. Nếu kệ B quá lớn thì hết chỗ đặt bản photo.

Partitioned hash: Chia cả hai kệ thành N ngăn theo ký tự đầu mã. Ngăn A1 chỉ join với ngăn B1, ngăn A2 chỉ với B2 — độc lập. Nếu mã "P001" có 90% đơn hàng (hot key), ngăn đó vẫn quá tải dù các ngăn khác rảnh.

2. Sort-merge join — sort một lần, merge mãi mãi

Sort-merge join là chiến lược mặc định của MapReduce và Spark khi cả hai bảng lớn. Tư tưởng đơn giản: nếu cả hai bảng đã được sort theo cùng join key, việc join chỉ là chạy merge pass một lần qua cả hai.

2.1 Cơ chế

Buoc 1 — Shuffle (phat ca hai bang cung key):
  map(event):   emit(product_id, ("E", event))   -- phat theo join key, tag loai
  map(product): emit(product_id, ("P", product)) -- ca hai bang cung key

  -> shuffle gom moi record co cung product_id ve cung reducer
  -> framework sort theo product_id truoc khi vao reduce

Buoc 2 — Merge (trong reducer):
  reduce(product_id, [("P", product), ("E", event1), ("E", event2), ...]):
      for each event in events:
          emit(join(event, product))   -- product da co san cung partition

Time:  O(n log n) shuffle+sort + O(n+m) merge
Space: O(1) per reducer -- streaming merge, khong load toan bo vao RAM

Hadoop gọi kỹ thuật gom cả hai bảng về cùng reducer theo join key là secondary sort: composite key (product_id, table_tag) đảm bảo bản ghi từ bảng phụ (products) luôn tới trước bản ghi sự kiện, nên reducer build lookup trong khi stream sự kiện chạy qua — không cần buffer toàn bộ.

2.2 Khi nào dùng

• Cả hai bảng lớn (GB–TB) — không fit broadcast.
• Dữ liệu có skew (hot key): sort-merge gom mỗi key về reducer riêng, reducer nào xử lý đúng phần của mình — không có reducer "nhận toàn đống" như hash join.
• Dữ liệu đã được sort sẵn từ bước trước → sort-merge "miễn phí" (chi phí sort gần như 0).

Chi phí chính là hai lần sort đầy đủ. Đổi lại: không cần RAM lớn tại reducer (streaming merge), và không bị phá vỡ bởi hot key skew.

Shuffle trong MapReduce (bài 02 — MapReduce) chính là bước sort này — framework sort record theo key trước reducer, nên sort-merge join là kết quả tự nhiên: bạn emit cùng join key từ cả hai bảng, framework lo phần còn lại.

Bước merge — k-way merge nhiều sorted stream — dùng cùng ý tưởng với merge sort đã học: N sorted run, merge đồng thời, O(n log k) với k là số sorted run merge cùng lúc (fan-in). Trong batch phân tán, k-way merge chạy song song trên nhiều reducer, mỗi reducer merge các run của partition mình.

3. Broadcast hash join — bơm bảng nhỏ vào RAM

Broadcast hash join tránh hoàn toàn việc shuffle bảng lớn. Thay vào đó, bảng nhỏ được copy lên RAM của mọi executor, rồi mỗi executor tự probe hash table cục bộ khi scan bảng lớn.

3.1 Cơ chế

Buoc 1 — Broadcast (driver doc bang nho):
  Driver load toan bo products vao RAM
  -> serialize + gui sang moi executor (broadcast variable)
  -> moi executor build: HashMap<product_id, product>

Buoc 2 — Probe (map-side, khong co shuffle phase):
  map(event):
      product <- localHashTable.get(event.product_id)   -- O(1), in RAM
      if product != null:
          emit(join(event, product))

Khong co reduce phase! Bang lon (events) khong di chuyen qua mang.

Time:  O(m) build hash table + O(n) probe, chay song song tren moi executor
Space: hash table phai fit vao RAM cua tung executor (khong chia se)

Spark tự chọn broadcast hash join khi Spark ước lượng bảng nhỏ hơn spark.sql.autoBroadcastJoinThreshold — mặc định 10 MB. Hard limit kỹ thuật là 8 GB (trên 8 GB Spark ném exception; ngay cả dưới limit, kích thước thực tế nhân số executor là gánh nặng RAM thật sự).

3.2 Khi nào dùng

• Bảng nhỏ thực sự nhỏ — safe zone: dưới vài chục MB, không phải GB-scale.
• Muốn tránh shuffle bảng lớn: bảng lớn không di chuyển qua mạng, tiết kiệm I/O.
• Join với dimension table cố định: mã quốc gia, danh mục, trạng thái.

Pitfall chính về kích thước sẽ nói kỹ ở Section 6.

4. Partitioned hash join — chia để trị

Partitioned hash join (Spark gọi là shuffle hash join) chia cả hai bảng theo cùng một hàm hash trên join key, rồi join từng cặp partition độc lập. Ý tưởng: nếu hash(product_id) % N cho cùng kết quả trên cả hai bảng, record cùng key sẽ vào cùng partition — và mỗi partition join độc lập mà không cần toàn bộ bảng nhỏ trên RAM.

4.1 Cơ chế

Buoc 1 — Partition (shuffle ca hai bang cung ham hash):
  map(event):   emit(hash(product_id) % N, event)
  map(product): emit(hash(product_id) % N, product)
  -> cung hash value -> cung partition -> cung reducer

Buoc 2 — Build + Probe (trong tung partition):
  reduce(partition_id, [events..., products...]):
      localHash <- build(products trong partition nay)  -- chi 1/N cua bang nho
      for each event in events:
          emit(join(event, localHash.get(event.product_id)))

Time:  O(n+m) build + probe, song song N partition
Space: RAM = kich thuoc 1 partition cua bang nho (khong phai toan bo)

Ưu điểm so với broadcast: mỗi reducer chỉ giữ 1/N của hash table — bảng nhỏ không cần fit toàn bộ vào một nơi. Ưu điểm so với sort-merge: không cần sort, chỉ hash, thường nhanh hơn khi dữ liệu phân phối đều.

4.2 Khi nào dùng

• Cả hai bảng lớn nhưng phân phối key đều (không hot key nghiêm trọng).
• Bảng nhỏ không đủ nhỏ để broadcast, nhưng 1/N của nó vừa RAM reducer.
• Không cần sort (partitioned hash join thường nhanh hơn sort-merge khi dữ liệu đều).

Nhược điểm nặng: hot key skew dồn một partition thành khổng lồ → straggler → phá vỡ toàn bộ ưu điểm song song.

5. Khi nào chọn chiến lược nào?

AQE (Adaptive Query Execution) là cơ chế Spark tối ưu lại kế hoạch join tại runtime dựa trên kích thước partition đo được thật (thay vì chỉ ước lượng tĩnh trước khi chạy). Nếu sau các bước lọc một bảng hoá ra dưới broadcast threshold → đổi sang broadcast; nếu chỉ đủ nhỏ để build hash mỗi partition (nhưng trên threshold) → đổi sang partitioned/shuffle hash.

Đáp án cho ba scenario:

1. Broadcast — 200 KB vừa RAM mọi executor, không shuffle bảng events 2 TB. Đây là trường hợp lý tưởng cho broadcast.
2. Partitioned hash — hai bảng lớn, phân phối đều → hash join nhanh hơn sort; không cần sort O(n log n).
3. Sort-merge — user_id skew cao, sort-merge không bị straggler (mỗi reducer đúng key của mình). Hoặc partitioned hash + salting (Section 6.2).

6. Pitfall — hai bẫy hay gặp

6.1 Broadcast bảng "nhỏ" vượt RAM

❌ Nhầm: "bảng products 8 GB nhỏ hơn bảng events 2 TB — broadcast là nhanh hơn."

SAI:
  products = 9 GB (bi ep BROADCAST hint)
  50 executors, moi executor giu 1 ban doc lap: 50 x 9 GB = 450 GB RAM
  9 GB > hard limit 8 GB -> SparkException: broadcast size exceeds limit
  (ngan cach nay la ky thuat; ngay ca 1 GB tren 50 executor = 50 GB van la qua tai RAM)

✅ Đúng:

DUNG:
  products 8 GB >> broadcast threshold (10 MB mac dinh)
  -> dung sort-merge join hoac partitioned hash join
  Broadcast chi an toan khi bang nho thuc su vua RAM moi executor
  Safe zone thuc te: vai chuc MB, khong phai GB-scale

Cách tính nhanh trước khi broadcast: kích_thước_bảng × số_executor ≤ RAM khả dụng per executor. Ngưỡng 10 MB mặc định không phải con số ngẫu nhiên — đó là ước tính dè dặt để bảng broadcast không chiếm quá vài phần trăm heap executor thông thường.

6.2 Reducer skew từ hot key

❌ Nhầm: "hash(product_id) % N chia đều công việc cho N reducer."

Hash chia đều số key khác nhau, KHÔNG chia đều số record. Nếu product_id = "P001" xuất hiện trong 50 triệu dòng log (sản phẩm hot), toàn bộ 50 triệu dòng đó vào một reducer — trong khi hàng nghìn key khác mỗi cái vài trăm dòng.

Một reducer có thể nhận gấp hàng trăm tới hàng nghìn lần record so với mức trung bình (tuỳ độ nghiêng power-law của join key) — job bị straggler, toàn pipeline đứng chờ.

SAI: hash(product_id) % N — khong co xu ly hot key
  "P001" -> hash -> partition 7 -> 50 trieu record (straggler)
  "P002" -> hash -> partition 3 -> 200 record
  "P003" -> hash -> partition 1 -> 150 record
  ... 998 partition khac xong trong 30 giay, partition 7 mat 45 phut ...

TODO: them salting cho hot key P001 truoc khi emit

✅ Fix: salted join

DUNG — salted join:
  map(event):
      salt <- random(0..N)
      emit((event.product_id + "_" + salt), event)   -- trai sang N partition

  -- bang phu (products) phai replicate N lan de match salt:
  map(product):
      for salt in 0..N-1:
          emit((product.id + "_" + salt), product)   -- replicate

  reduce((product_id + salt), [events..., product]):
      -- moi partition nho, join nhanh; ket qua gop lai sau

Trade-off: bang phu tang N lan kich thuoc.
  N=10, products 1 GB -> 10 GB tong cong.
  Can so voi loi ich: neu straggler keo job tu 10 phut len 3 gio, xung dang.

7. 📚 Deep Dive

8. Liên hệ các bài khác

• Bài 02 — MapReduce: shuffle trong MapReduce chính là bước sort của sort-merge join — framework sort record theo key trước reducer, nên emit cùng join key từ hai bảng là đủ để join.
• Bài 04 — Dataflow và fault tolerance: trong pipeline join nhiều bước, materialization và pipelining quyết định bao nhiêu dữ liệu ghi xuống disk giữa các bước — fault tolerance khi một bước join thất bại.
• Merge sort — k-way merge: bước merge trong sort-merge join chính là k-way merge nhiều sorted stream đã học — N sorted partition, merge đồng thời, O(n log k) với k là số partition.
• External sort — sắp xếp ngoài RAM: shuffle của MapReduce thực chất là external sort phân tán — chunk-sort-merge khi dữ liệu lớn hơn RAM của từng máy.

9. Tóm tắt

• Bẫy broadcast: tính bytes × executor count trước — 8 GB lên 50 executor là 400 GB RAM. Ngưỡng mặc định Spark 10 MB, hard limit 8 GB.
• Khi skew: hash(key) % N chia đều số key, không đều record. Fix bằng salted join — replicate bảng phụ N lần; trade-off: bảng phụ tăng N lần kích thước.
• Nguyên tắc chọn: kích thước quyết định broadcast vs phần còn lại; skew quyết định sort-merge vs hash.
• Sort-merge join là fallback mặc định (không bị skew, terabyte OK); broadcast hash join khi bảng nhỏ thực sự nhỏ; partitioned hash join khi dữ liệu đều và tránh sort.

10. Tự kiểm tra

Bài tiếp theo: Dataflow engine — DAG, materialization và fault tolerance