Spark chọn chiến lược join thế nào? Broadcast hash join và sort-merge join khác nhau ra sao?

Với join theo điều kiện bằng (equi-join), Spark chủ yếu chọn giữa các cách sau, dựa trên kích thước ước lượng của hai bảng:

• Broadcast hash join: bảng nhỏ được gửi nguyên bản tới mọi executor, mỗi partition của bảng lớn join tại chỗ. Bảng lớn không phải shuffle nên đây là cách nhanh nhất. Spark tự chọn khi một bên nhỏ hơn spark.sql.autoBroadcastJoinThreshold (mặc định 10MB).
• Sort-merge join: cả hai bảng được shuffle theo key join, sort trong từng partition rồi duyệt song song để ghép. Chịu được hai bảng đều lớn vì không phải giữ bảng nào trọn trong RAM, đổi lại tốn shuffle và sort. Đây là mặc định khi cả hai bên đều lớn.
• Shuffle hash join: shuffle cả hai bên nhưng dựng hash table từ bên nhỏ hơn trong từng partition, bỏ bước sort. Hợp khi một bên nhỏ hơn hẳn nhưng chưa đủ nhỏ để broadcast.
• Join không phải điều kiện bằng (a.ts BETWEEN b.start_ts AND b.end_ts) rơi vào broadcast nested loop join hoặc cartesian product, chậm hơn nhiều.

from pyspark.sql.functions import broadcast

orders.join(broadcast(dim_store), "store_id")

SELECT /*+ BROADCAST(s) */ o.*, s.region
FROM orders o JOIN dim_store s ON o.store_id = s.store_id;

Với AQE (bật mặc định từ Spark 3.2), Spark còn đổi sort-merge join sang broadcast join lúc chạy nếu sau khi lọc một bên thực tế nhỏ hơn ngưỡng. Xem plan thực tế bằng df.explain() hoặc tab SQL trên Spark UI, tìm BroadcastHashJoin hay SortMergeJoin.

Lưu ý: broadcast bảng quá lớn làm driver hoặc executor hết bộ nhớ, vì bảng phải gom về driver rồi nhân bản lên từng executor. Chỉ tăng ngưỡng broadcast khi đã đo bộ nhớ. Ngược lại, bảng dimension lớn dần theo thời gian có thể vượt ngưỡng, và job lặng lẽ chuyển sang sort-merge join rồi chậm hẳn đi.