Spark 5 — Tối ưu hiệu năng & bộ nhớ

Vì sao cần tuning?

Một Spark job "chạy được" và một Spark job "chạy tốt" là hai chuyện hoàn toàn khác nhau. Job đầu tiên bạn viết thường chạy được — nhưng khi dữ liệu lớn dần lên, cùng đoạn code ấy bắt đầu chậm rề rề, rồi một ngày đẹp trời văng thẳng lỗi OutOfMemoryError và cả cluster đứng hình. Tuning là nghệ thuật (và một phần khoa học) làm cho job vừa nhanh, vừa ổn định, vừa tiết kiệm tài nguyên.

Điều quan trọng nhất cần khắc cốt ghi tâm ngay từ đầu: tuning bắt đầu bằng đo, không bằng đoán. Rất nhiều người vừa thấy job chậm là lao vào tăng spark.executor.memory gấp đôi, hoặc rải .cache() khắp nơi. Phần lớn những thao tác đó vô ích, thậm chí phản tác dụng. Cách làm đúng là mở Spark UI, tìm ra nút thắt lớn nhất (bottleneck), rồi chỉ sửa đúng chỗ đó. Bài này đi qua từng vũ khí trong kho: mô hình bộ nhớ, kích cỡ executor, cache, broadcast join, partition, AQE, và cách chẩn đoán spill/OOM.

Bài này nối tiếp trực tiếp Shuffle & Partitions — nếu chưa nắm shuffle, bạn nên đọc bài đó trước, vì phần lớn vấn đề hiệu năng của Spark quy về shuffle.

Lưu ý về môi trường: Sandbox của loạt bài này dùng PostgreSQL, nên các đoạn PySpark bên dưới không chạy trực tiếp ở đây — chúng là minh hoạ để bạn mang lên cluster thật. Mọi con số cấu hình đều theo Spark 3.5+.

Mô hình bộ nhớ của executor

Muốn hiểu vì sao job OOM, phải hiểu bộ nhớ một executor được chia ra sao. Mỗi executor là một tiến trình JVM, và RAM cấp cho nó (spark.executor.memory) được JVM chia thành các vùng.

Có mấy tầng quan trọng:

• Reserved memory (~300 MB): phần Spark giữ lại cho hoạt động nội bộ của chính nó. Bạn không đụng tới được.
• User memory (~40% phần còn lại): dùng cho các cấu trúc dữ liệu người dùng tạo ra trong code (biến trong UDF, mảng, hashmap tự viết). Đây là lý do UDF phức tạp dễ ngốn bộ nhớ.
• Unified memory (~60%, do spark.memory.fraction = 0.6 điều khiển): vùng lớn nhất, được chia tiếp thành hai nửa có thể vay mượn lẫn nhau:
  • Execution memory: dùng cho shuffle, join, sort, aggregation — tức những phép tính "nặng".
  • Storage memory: dùng để giữ dữ liệu cache()/persist() và các bảng broadcast.

Điểm tinh tế của mô hình unified memory (từ Spark 1.6) là ranh giới giữa execution và storage không cố định. Khi execution cần thêm mà storage đang trống, nó mượn. Ngược lại storage có thể mượn vùng execution đang rảnh. Nhưng có một luật bất đối xứng quan trọng: execution được quyền "đòi lại" bộ nhớ đã cho storage mượn (đẩy block cache ra đĩa hoặc xoá), còn storage thì không được đòi lại phần execution đang dùng. Lý do: mất một block cache chỉ tốn công tính lại; còn cắt ngang một phép shuffle giữa chừng thì hỏng cả job. Đây chính là vì sao cache "bừa" đôi khi vô dụng — block bạn cache bị đá ra ngay khi có phép join cần chỗ.

Overhead memory và OOM đến từ đâu

Ngoài heap JVM còn có overhead memory (spark.executor.memoryOverhead) — vùng bộ nhớ ngoài heap, dùng cho buffer mạng, netty, thread stack, và đặc biệt là PySpark: khi bạn chạy Python UDF, dữ liệu phải serialize qua tiến trình Python nằm ngoài JVM, và bộ nhớ đó tính vào overhead chứ không vào heap. Mặc định overhead là max(384 MB, 10% executor.memory).

Bốn nguồn OOM phổ biến nhất:

1. OOM heap (java.lang.OutOfMemoryError: Java heap space): một task xử lý partition quá lớn (thường do skew) hoặc bạn collect() quá nhiều dữ liệu.
2. OOM overhead ("Container killed by YARN for exceeding memory limits"): overhead quá nhỏ, hay gặp với PySpark UDF hoặc nhiều partition nhỏ. Sửa bằng tăng memoryOverhead.
3. OOM driver: do collect() / toPandas() kéo bộ dữ liệu lớn về driver (xem phần dưới).
4. GC quá nhiều: heap gần đầy, JVM dành phần lớn thời gian gom rác thay vì tính toán — job không chết nhưng chậm kinh khủng.

Kích cỡ executor & core

Giả sử cluster của bạn có các node 16 core / 64 GB RAM. Bạn cấu hình executor to hay nhỏ? Có hai thái cực:

• Fat executor (ví dụ 1 executor = 16 core, ~60 GB): ít tiến trình JVM, chia sẻ broadcast/cache tốt trong một executor. Nhưng nhiều core dùng chung một heap gây áp lực GC lớn, và một executor chết là mất nhiều task cùng lúc. HDFS I/O cũng bão hoà khi quá nhiều luồng đọc/ghi cùng lúc.
• Thin executor (ví dụ 1 executor = 1 core, ~4 GB): cách ly tốt, GC nhẹ. Nhưng lãng phí overhead (mỗi executor tốn ~300 MB reserved + overhead riêng), và mất khả năng chia sẻ broadcast/cache — mỗi executor phải giữ bản sao riêng.

Kinh nghiệm cân bằng đã được cộng đồng đúc kết: ~4–5 core mỗi executor. Đây là điểm ngọt vì trên 5 core/executor thì thông lượng HDFS bắt đầu suy giảm, còn dưới đó thì lãng phí. Cách tính thực tế:

(minh hoạ) Node 16 core, 64 GB
- Chừa 1 core + 1 GB cho OS/daemon  -> còn 15 core
- Chọn 5 core/executor               -> 3 executor/node
- RAM mỗi executor ~ (64-1)/3 ≈ 21 GB
  -> executor.memory ≈ 18 GB, memoryOverhead ≈ 2-3 GB

Từ đó suy ra độ song song tổng = (số executor) × (số core mỗi executor). Ví dụ 10 node × 3 executor × 5 core = 150 task chạy song song cùng lúc. Con số này là chìa khoá liên kết với partition: nếu dữ liệu chỉ có 40 partition thì 110 core còn lại ngồi chơi; nếu có 15.000 partition tí hon thì overhead lập lịch task áp đảo. Quy tắc ngón tay cái: số partition nên gấp 2–4 lần tổng số core để lấp đầy và cân tải khi có task lệch giờ. Xem thêm cách chỉnh partition ở Shuffle & Partitions.

Cache / persist đúng cách

Spark tính lazy: mỗi action (như count(), write()) tính lại toàn bộ chuỗi transformation từ đầu. Nếu bạn dùng lại cùng một DataFrame nhiều lần, Spark sẽ tính đi tính lại — lãng phí. Đó là lúc cache có ý nghĩa.

# (minh hoạ)
df_clean = (spark.read.parquet("s3://.../events")
            .filter("status = 'ACTIVE'")
            .join(dim, "user_id"))

# df_clean sẽ được dùng cho NHIỀU output khác nhau
df_clean.cache()          # = persist(MEMORY_AND_DISK) trong DataFrame API
df_clean.count()          # action "vật chất hoá" cache

report_a = df_clean.groupBy("country").count()
report_b = df_clean.groupBy("device").sum("amount")
# cả hai đọc từ cache, không tính lại join

df_clean.unpersist()      # GIẢI PHÓNG khi dùng xong

Vài nguyên tắc:

• Chỉ cache khi tái dùng ≥ 2 lần. Cache một DataFrame chỉ dùng một lần là phản tác dụng: tốn công ghi cache mà chẳng bao giờ đọc lại, còn chiếm chỗ storage memory của người khác.
• Chọn StorageLevel hợp lý. MEMORY_ONLY nhanh nhất nhưng nếu không đủ RAM thì block bị bỏ (phải tính lại). MEMORY_AND_DISK (mặc định của .cache() ở DataFrame API) an toàn hơn: hết RAM thì tràn xuống đĩa. Còn MEMORY_ONLY_SER / DISK_ONLY đánh đổi CPU serialize lấy chỗ.
• Luôn unpersist() khi không cần nữa — đừng để cache cũ ăn chỗ của các stage sau.
• Đừng cache dữ liệu vừa lọc ít. Nếu chi phí tính lại rẻ (chỉ vài filter trên parquet đã pushdown), cache có khi còn chậm hơn.

Một cạm bẫy hay gặp: cache trước khi lọc/chiếu cột, khiến bạn nhét cả đống dữ liệu thô vào bộ nhớ. Hãy select và filter trước, cache sau — chỉ giữ đúng phần bạn thật sự tái dùng.

Broadcast join — vũ khí chống shuffle

Shuffle là kẻ thù số một của hiệu năng. Khi join hai bảng lớn, Spark phải xáo trộn (shuffle) cả hai theo key qua mạng — rất tốn kém. Nhưng nếu một bảng đủ nhỏ, có một mẹo: gửi (broadcast) nguyên bảng nhỏ đó tới mọi executor, rồi join cục bộ tại chỗ. Không cần shuffle bảng lớn nữa. Đây gọi là broadcast hash join (hay map-side join).

Spark tự động broadcast khi ước lượng kích thước bảng nhỏ hơn spark.sql.autoBroadcastJoinThreshold, mặc định 10 MB (10485760 bytes). Nhưng việc "tự động" phụ thuộc vào Spark ước lượng đúng kích thước — với file parquet đã nén hay bảng chưa có thống kê, ước lượng thường sai và Spark bỏ qua broadcast. Khi đó bạn ép thủ công bằng hint:

# (minh hoạ)
from pyspark.sql.functions import broadcast

# dim_country nhỏ (vài nghìn dòng) -> broadcast, tránh shuffle fact
result = big_fact.join(broadcast(dim_country), "country_code")

# Hoặc nâng ngưỡng tự động (cẩn thận, đừng để quá lớn)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 50 * 1024 * 1024)  # 50 MB

# Tắt hẳn broadcast (khi cần) = -1
# spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)

Lưu ý quan trọng: bảng broadcast phải nằm gọn trong bộ nhớ driver (nơi gom lại trước khi phát) và trong storage memory của mỗi executor. Broadcast một bảng "nhỏ" nhưng thực ra 2 GB sẽ làm OOM driver. Quy tắc an toàn: chỉ broadcast bảng chiều (dimension) cỡ vài chục MB trở xuống. Broadcast join cũng là một trong ba tối ưu mà AQE có thể tự áp dụng lúc runtime.

Partition & AQE

Số partition sau shuffle do spark.sql.shuffle.partitions quyết định — mặc định 200. Con số 200 này gần như luôn sai với dữ liệu thật: quá nhiều cho job nhỏ (200 task tí hon, overhead lập lịch), quá ít cho job lớn (mỗi partition khổng lồ, spill và OOM). Trước đây bạn phải tự chỉnh tay theo từng job.

Adaptive Query Execution (AQE) giải quyết chuyện này. Từ Spark 3.2 AQE được bật mặc định (spark.sql.adaptive.enabled = true). AQE dùng thống kê runtime (kích thước thật sau khi shuffle) để tối ưu lại kế hoạch, với ba khả năng chính:

• Coalesce partitions: gộp các partition nhỏ sau shuffle lại, để bạn không còn phải căn shuffle.partitions thủ công. Bạn có thể đặt số cao (ví dụ 400–800) rồi để AQE tự gộp xuống.
• Chuyển sort-merge join thành broadcast join: nếu runtime phát hiện một nhánh join thực ra đủ nhỏ, AQE tự broadcast — kể cả khi ước lượng ban đầu sai.
• Skew join optimization: phát hiện partition lệch (skew) và tách nó ra thành nhiều task nhỏ đều nhau. Bật kèm spark.sql.adaptive.skewJoin.enabled (mặc định true). Một partition bị coi là skew khi vượt skewedPartitionThresholdInBytes (mặc định 256 MB) và lớn hơn skewedPartitionFactor (mặc định 5.0) lần trung vị.

# (minh hoạ) cấu hình tuning điển hình cho một job vừa
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.shuffle.partitions", "400")   # AQE sẽ tự gộp bớt

AQE mạnh nhưng không phải thuốc tiên — nó tối ưu trong kế hoạch đã sinh, không sửa được thiết kế join tồi hay skew do chính khoá dữ liệu (ví dụ 80% bản ghi có user_id = NULL). Với skew "bệnh lý", vẫn cần kỹ thuật thủ công như salting. Chi tiết ở Shuffle & Partitions.

Spill & OOM — chẩn đoán qua Spark UI

Spill xảy ra khi một task cần nhiều execution memory hơn nó có (thường trong sort/aggregate/join), buộc phải ghi tạm dữ liệu xuống đĩa rồi đọc lại. Spill không làm chết job nhưng chậm khủng khiếp vì đĩa chậm hơn RAM hàng trăm lần. Trên Spark UI, vào tab Stages, xem cột "Spill (Memory)" và "Spill (Disk)" — nếu chúng lớn, đó là dấu hiệu rõ ràng.

Cách giảm spill:

• Tăng số partition (giảm kích thước mỗi partition, mỗi task nhẹ hơn) — thường là cách rẻ nhất.
• Tăng execution memory (executor.memory lớn hơn, hoặc giảm số core/executor để mỗi core có nhiều RAM hơn).
• Giảm dữ liệu phải xử lý cùng lúc: lọc sớm, chiếu cột sớm.

Chẩn đoán qua UI theo trình tự:

1. Tab Jobs/Stages: stage nào chạy lâu nhất? Đó là nút thắt.
2. Event Timeline & Task distribution: trong stage đó, nếu một vài task chạy lâu hơn hẳn (max duration ≫ median), đó là skew.
3. Cột Spill: có spill lớn không? -> thiếu memory/partition.
4. Tab Storage: cache có thật sự được dùng không, tỉ lệ "Fraction cached" bao nhiêu?
5. Tab Executors: cột GC Time — nếu GC chiếm >10% task time, heap thiếu hoặc cache quá nhiều.

Về OOM driver: tránh collect(), toPandas(), take(n) với n lớn trên dữ liệu lớn — chúng kéo toàn bộ về một máy driver. Nếu chỉ cần xem thử, dùng show(20) hoặc limit(). Muốn xuất ra ngoài thì write xuống storage phân tán, đừng gom về driver. Cũng cẩn thận với explode() trên mảng lớn: nó nhân số dòng lên theo cấp số nhân và dễ làm nổ partition.

Định dạng & I/O

Tối ưu tính toán mà đọc/ghi cẩu thả thì vẫn chậm. Vài nguyên tắc I/O:

• Dùng Parquet (cột) + nén (Snappy mặc định, hoặc zstd). Parquet cho phép hai tối ưu lớn mà CSV/JSON không có:
  • Column pruning: select cột nào thì chỉ đọc cột đó khỏi đĩa.
  • Predicate pushdown: điều kiện filter được đẩy xuống tận tầng đọc file, bỏ qua cả row group không thoả.
• Partition pruning: nếu dữ liệu được ghi phân vùng theo thư mục (ví dụ .../dt=2026-06-30/), một filter dt = '2026-06-30' chỉ đọc đúng thư mục đó, bỏ qua phần còn lại. Đây là một trong những tối ưu I/O đáng giá nhất cho bảng lớn theo ngày.
• Tránh "small files problem": hàng nghìn file vài KB làm Spark tốn công mở/đóng file và sinh vô số task tí hon. Ghi ra bằng .coalesce(n) hoặc .repartition(n) để gộp thành file cỡ 128 MB – 1 GB. Với bảng đích lâu dài, chạy job "compaction" định kỳ để gộp file nhỏ.

Use case thực tế: job OOM & chậm

Bối cảnh. Một pipeline hằng đêm join bảng transactions (500 GB) với dim_merchant (30 MB) rồi tổng hợp doanh thu theo merchant. Gần đây job chạy từ 25 phút vọt lên 2 tiếng, thỉnh thoảng văng Container killed by YARN... exceeding memory limits.

Chẩn đoán qua Spark UI.

1. Tab Stages: stage join chiếm 90% thời gian. Kế hoạch (tab SQL) cho thấy Spark dùng SortMergeJoin — tức đang shuffle cả 500 GB, dù dim_merchant chỉ 30 MB. Nguyên nhân: bảng lưu parquet nén, Spark ước lượng kích thước sai nên bỏ qua broadcast (mặc định ngưỡng chỉ 10 MB, còn 30 MB thì vượt).
2. Tab Executors: một số task trong stage aggregate có max duration gấp ~40 lần median — dấu hiệu skew: một số merchant lớn chiếm phần lớn giao dịch.
3. Cột Spill (Disk) rất lớn ở stage aggregate — thiếu execution memory do partition khổng lồ.
4. YARN kill là do overhead: PySpark UDF chuẩn hoá tên merchant đang ngốn bộ nhớ ngoài heap.

Sửa (theo thứ tự tác động lớn nhất trước).

# (minh hoạ)
from pyspark.sql.functions import broadcast

# 1) Ép broadcast bảng nhỏ -> loại shuffle 500 GB (thắng lớn nhất)
res = transactions.join(broadcast(dim_merchant), "merchant_id")

# 2) Bật/để AQE xử lý skew ở stage aggregate
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# 3) Tăng partition để giảm spill
spark.conf.set("spark.sql.shuffle.partitions", "600")

# 4) Tăng overhead cho PySpark UDF (thay vì tăng cả executor.memory)
#    --conf spark.executor.memoryOverhead=3g

# 5) Nếu UDF chỉ chuẩn hoá chuỗi -> thay bằng hàm built-in (Catalyst tối ưu được)
#    trim(lower(col("name")))  nhanh hơn Python UDF nhiều

Kết quả (điển hình cho loại sửa này). Loại được shuffle 500 GB nhờ broadcast là thắng lợi lớn nhất; AQE tách partition skew làm các task đều nhau, spill giảm; đổi UDF sang built-in gỡ luôn áp lực overhead nên hết bị YARN kill. Bài học xuyên suốt: đo trước, sửa nút thắt lớn nhất trước — chỉ riêng bước broadcast đã giải quyết phần lớn vấn đề, không cần tăng RAM bừa. Các thao tác vận hành cluster đầy đủ hơn nằm ở Production.

Ghi nhớ

• Đo trước, đoán sau. Mở Spark UI, tìm stage lâu nhất, sửa đúng nút thắt. Đừng tăng RAM hay cache theo cảm tính.
• Bộ nhớ executor chia thành reserved / user / unified (execution ↔ storage vay mượn động). OOM đến từ heap (skew, collect), overhead (PySpark, nhiều partition), hoặc driver (collect).
• Executor ~4–5 core, chừa tài nguyên cho OS. Độ song song = executors × cores; đặt số partition ~2–4× tổng core.
• Cache chỉ khi tái dùng ≥ 2 lần, chọn MEMORY_AND_DISK cho an toàn, và luôn unpersist(). Lọc/chiếu cột trước khi cache.
• Broadcast bảng nhỏ (vài chục MB) để loại shuffle bảng lớn; dùng hint broadcast() khi ước lượng của Spark sai. Đừng broadcast bảng to (OOM driver).
• Bật AQE (mặc định từ 3.2): coalesce partition, skew join, tự chuyển sang broadcast. Đặt shuffle.partitions rộng rồi để AQE gộp.
• Spill/GC đọc từ Spark UI: giảm bằng tăng partition/memory; tránh collect() và explode() quá mức.
• Parquet + nén, pushdown, partition pruning, tránh small files (gộp về 128 MB–1 GB).
• Đọc cùng: Shuffle & Partitions, Spark SQL, Production.