Spark 4 — Shuffle, partition & data skew

Vì sao bài này quan trọng nhất trong loạt Spark

Nếu bạn chỉ có thời gian đọc một bài duy nhất về tối ưu Spark, hãy đọc bài này. Trong thực tế vận hành, phần lớn các job Spark chạy chậm, treo, hoặc "Out Of Memory" đều quy về một nguyên nhân: shuffle. Một job xử lý vài trăm GB có thể chạy trong 3 phút hoặc trong 3 tiếng, tùy vào việc bạn hiểu shuffle và data skew đến đâu.

Ở bài Spark SQL, ta đã thấy Catalyst tối ưu logical plan và Tungsten tối ưu thực thi. Nhưng có một chi phí mà không optimizer nào xóa bỏ được hoàn toàn: khi dữ liệu phải di chuyển giữa các máy qua mạng. Bài này giải phẫu chính xác chi phí đó, và cách kiểm soát nó.

Partition: đơn vị song song của Spark

Trước khi nói về shuffle, phải hiểu partition. Một DataFrame trong Spark không nằm ở một chỗ — nó bị chia thành nhiều mảnh nhỏ gọi là partition, mỗi partition là một tập con các dòng nằm trên một executor.

Partition là đơn vị song song hóa cơ bản. Quy tắc vàng:

Một partition = một task = một core xử lý tại một thời điểm.

Nếu DataFrame của bạn có 200 partition và cluster có 100 core, Spark chạy được 100 task cùng lúc, còn 100 task chờ. Nếu DataFrame chỉ có 4 partition nhưng cluster có 100 core, thì 96 core ngồi chơi — bạn trả tiền cho 100 máy nhưng chỉ dùng 4.

Chọn số partition là bài toán cân bằng:

• Quá ít partition → không tận dụng hết cluster; mỗi partition quá lớn dễ tràn bộ nhớ (spill), task chạy lâu.
• Quá nhiều partition → mỗi task xử lý quá ít dữ liệu, nhưng chi phí khởi tạo/lập lịch/quản lý mỗi task (overhead) trở nên đáng kể. Hàng trăm nghìn task tí hon còn làm nghẽn driver.

Mục tiêu thực dụng: kích thước mỗi partition sau shuffle nằm trong khoảng ~128MB đến ~200MB. Đây là "con số cảm giác" mà cộng đồng Spark hay dùng, đủ lớn để amortize overhead nhưng đủ nhỏ để vừa bộ nhớ và cân bằng tải.

Narrow vs Wide transformation

Đây là khái niệm chia đôi thế giới transformation của Spark, và là chìa khóa để hiểu shuffle.

Narrow transformation

Mỗi partition đầu ra chỉ phụ thuộc vào một partition đầu vào. Dữ liệu không di chuyển giữa các máy. Mỗi executor xử lý phần dữ liệu đang nằm sẵn trên nó.

Ví dụ: map, filter, select, withColumn. Khi bạn filter một DataFrame, mỗi executor lọc phần dữ liệu của riêng mình, không cần hỏi executor khác. Cực rẻ, chạy song song hoàn toàn.

Wide transformation

Mỗi partition đầu ra phụ thuộc vào nhiều partition đầu vào, thường là tất cả. Để tính được kết quả, Spark phải gom các dòng có cùng key về cùng một chỗ — và điều đó buộc dữ liệu di chuyển qua mạng. Đây chính là shuffle.

Ví dụ: groupBy, join, distinct, reduceByKey, orderBy, repartition. Khi bạn groupBy("customer_id"), mọi dòng của cùng một customer_id phải về chung một partition để cộng dồn — nhưng ban đầu chúng rải rác trên khắp cluster.

Ranh giới shuffle chính là ranh giới stage. Spark chia DAG thực thi thành các stage, cắt tại mỗi wide transformation. Nội trong một stage, các narrow transformation được "hợp nhất" (pipelining) và chạy liền mạch.

Vì sao shuffle đắt đến vậy

Shuffle không chỉ là "gửi dữ liệu qua mạng". Nó là một chuỗi thao tác nặng, mỗi thao tác đều tốn:

1. Ghi ra đĩa (shuffle write): Mỗi task ở stage trước tính hash(key) % số_partition_đích cho từng dòng, phân dòng vào các buffer, rồi ghi các file shuffle xuống đĩa cục bộ. Đây là I/O đĩa.
2. Truyền qua mạng: Các task ở stage sau kéo (fetch) các mảnh dữ liệu thuộc về mình từ đĩa của mọi executor khác. Đây là I/O mạng — thường là khâu chậm nhất.
3. Đọc lại và sắp xếp (shuffle read): Task ở stage sau đọc dữ liệu về, thường phải sắp xếp/merge (ví dụ sort-merge join) trước khi tính toán.

So sánh: đọc dữ liệu từ RAM tính bằng nanosecond, từ SSD cục bộ tính bằng chục microsecond, còn qua mạng data-center tính bằng trăm microsecond đến millisecond — chậm hơn RAM hàng nghìn lần. Nhân con số đó với hàng tỷ dòng, bạn hiểu vì sao shuffle luôn là nghi phạm số một khi job chậm.

Nguyên tắc thiết kế pipeline: giảm số lần shuffle và giảm lượng dữ liệu mỗi lần shuffle. Filter sớm, chọn cột cần thiết sớm (đẩy chúng lên trước wide transformation), tránh distinct/orderBy không cần thiết.

spark.sql.shuffle.partitions — con số 200 huyền thoại

Khi một shuffle xảy ra, Spark cần biết chia dữ liệu thành bao nhiêu partition đích. Với DataFrame/Spark SQL, con số đó do spark.sql.shuffle.partitions quyết định, mặc định là 200.

Con số 200 này là hằng số cố định, không phụ thuộc vào dữ liệu của bạn — và đó chính là vấn đề:

• Bạn shuffle 10 GB → 200 partition, mỗi partition ~50MB. Hơi nhỏ nhưng chấp nhận được.
• Bạn shuffle 2 TB → 200 partition, mỗi partition ~10GB. Thảm họa: mỗi task xử lý 10GB, chắc chắn tràn bộ nhớ, spill xuống đĩa liên tục, hoặc chết OOM.
• Bạn shuffle 500 MB → 200 partition, mỗi partition ~2.5MB. Quá nhỏ: 200 task tí hon, overhead lập lịch lấn át công việc thật.

Cách tính thủ công truyền thống:

# (minh hoạ) ước lượng số shuffle partition theo mục tiêu ~128MB/partition
tong_shuffle_bytes = 2 * 1024 * 1024 * 1024 * 1024   # ~2 TB
target_partition_bytes = 128 * 1024 * 1024           # 128 MB
so_partition = tong_shuffle_bytes // target_partition_bytes  # ~16000

spark.conf.set("spark.sql.shuffle.partitions", 16000)

Tin vui: từ Spark 3, AQE tự động làm việc này (xem phần cuối), nên bạn ít khi phải chỉnh tay shuffle.partitions nữa. Nhưng hiểu nó vẫn cần thiết vì AQE dùng nó làm điểm xuất phát trước khi coalesce.

Data skew: kẻ giết job thầm lặng

Giả sử bạn chỉnh số partition hoàn hảo, cluster khỏe, nhưng job vẫn treo ở 199/200 task xong còn 1 task chạy mãi. Đây gần như chắc chắn là data skew.

Data skew xảy ra khi dữ liệu phân bố không đều theo key. Shuffle phân dòng theo hash(key), nên nếu một key có quá nhiều dòng, tất cả dồn vào một partition → một task phải xử lý lượng dữ liệu khổng lồ trong khi các task khác đã xong từ lâu.

Ví dụ kinh điển: bảng giao dịch join với bảng khách hàng theo customer_id. Nếu 90% giao dịch thuộc về một tài khoản "hệ thống" (ví dụ tài khoản trung gian thanh toán), thì partition chứa key đó phình to gấp trăm lần các partition khác. Task xử lý nó trở thành straggler (kẻ chậm chân): một executor treo, cả job chờ nó, và nếu partition quá lớn thì executor đó OOM và cả job fail.

Nhận biết skew qua Spark UI

Trong tab Stages của Spark UI, mở stage đang treo và xem bảng thống kê task (Summary Metrics):

• Cột Duration: nếu Max lớn hơn Median hàng chục lần → task lệch.
• Cột Shuffle Read Size / Records: nếu Max lớn hơn Median rất nhiều → một partition nhận quá nhiều dữ liệu → skew.
• Biểu hiện điển hình: 199 task xong trong 30 giây, task thứ 200 chạy 45 phút.

Đối chiếu Max với Median (percentile 50) là cách chẩn đoán skew nhanh nhất. Nếu chúng gần nhau, dữ liệu cân bằng; nếu chênh lệch lớn, bạn có skew.

Xử lý skew

1. Salting (rắc muối): Thêm một hậu tố ngẫu nhiên vào key nóng để "bẻ" nó thành nhiều partition, rồi gộp lại sau. Đây là kỹ thuật kinh điển cho skew join khi bạn không thể dựa vào AQE.

# (minh hoạ) salting cho join bị skew
from pyspark.sql import functions as F

N = 10  # số "muối"

# Bảng lớn (giao dịch) — thêm salt ngẫu nhiên vào key nóng
tx_salted = tx.withColumn(
    "cust_salt",
    F.concat(F.col("customer_id"), F.lit("_"),
             (F.rand() * N).cast("int"))
)

# Bảng nhỏ (khách hàng) — nhân bản mỗi dòng thành N bản, mỗi bản một salt
salt_range = spark.range(N).withColumnRenamed("id", "salt")
cust_exploded = (cust
    .crossJoin(salt_range)
    .withColumn("cust_salt",
                F.concat(F.col("customer_id"), F.lit("_"), F.col("salt"))))

# Join theo key đã salt → dữ liệu key nóng trải đều trên N partition
result = tx_salted.join(cust_exploded, "cust_salt")

Salting biến một key nóng thành N key vừa phải, phá thế "dồn một chỗ". Cái giá là bảng nhỏ bị nhân bản N lần — chỉ đáng làm khi bảng đó thực sự nhỏ.

2. Tách key nóng (isolated join): Nếu chỉ vài key gây skew, tách chúng ra xử lý riêng: broadcast join phần key nóng (vì tập key nóng thường nhỏ), shuffle join phần còn lại, rồi union kết quả.

3. AQE skew join: Từ Spark 3, để AQE tự phát hiện và tách partition skew — thường là cách đầu tiên nên thử vì không phải sửa code (xem bên dưới).

repartition vs coalesce

Hai hàm này cùng đổi số partition nhưng khác nhau về bản chất và chi phí — nhầm lẫn chúng là lỗi phổ biến.

repartition(n) — có shuffle. Xáo trộn toàn bộ dữ liệu để chia lại thành n partition cân bằng đều. Dùng khi:

• Cần tăng số partition (coalesce không tăng được).
• Muốn dữ liệu cân bằng lại sau một filter mạnh (ví dụ lọc còn 1% dữ liệu nhưng vẫn giữ 200 partition rỗng/lệch).
• Muốn phân vùng theo cột để tối ưu join/ghi sau đó: repartition("customer_id").

coalesce(n) — không shuffle (khi giảm số partition). Chỉ gộp các partition hiện có trên cùng executor lại với nhau, tránh di chuyển dữ liệu qua mạng. Rẻ hơn nhiều, nhưng:

• Chỉ dùng để giảm số partition.
• Kết quả có thể không cân bằng (gộp partition lớn với nhỏ tùy hiện trạng).
• Nếu giảm quá mạnh (ví dụ từ 1000 xuống 1), coalesce có thể làm mất song song ở cả các stage trước đó.

# (minh hoạ) repartition vs coalesce
df.rdd.getNumPartitions()          # ví dụ: 200

# Giảm để ghi ra ít file — rẻ, không shuffle
df.coalesce(10).write.parquet("/out/report")

# Cân bằng lại sau filter mạnh — có shuffle
big = df.filter(F.col("status") == "ACTIVE")   # còn ~2% dữ liệu, 200 part lệch
big = big.repartition(20)

# Phân vùng theo cột trước khi ghi partitioned table
df.repartition("event_date").write.partitionBy("event_date").parquet("/out/events")

Quy tắc nhớ nhanh: giảm số file khi ghi → coalesce; cần cân bằng lại hoặc tăng partition → repartition.

Spill: khi partition không vừa bộ nhớ

Khi một task xử lý partition quá lớn (hoặc một thao tác như sort/aggregate cần nhiều bộ nhớ), Spark không có đủ RAM để giữ toàn bộ trong execution memory. Nó buộc phải spill — ghi dữ liệu trung gian xuống đĩa, rồi đọc lại. Spill làm chậm khủng khiếp vì biến thao tác trong RAM thành thao tác đĩa.

Trong Spark UI, cột Spill (Memory) và Spill (Disk) lớn là dấu hiệu partition quá to hoặc bộ nhớ quá nhỏ. Cách giảm spill:

• Tăng số shuffle partition để mỗi partition nhỏ lại (hoặc để AQE lo).
• Xử lý skew — spill thường tập trung ở đúng task skew.
• Tăng bộ nhớ executor, hoặc tăng tỉ lệ execution memory.
• Filter/select sớm để giảm lượng dữ liệu đi vào shuffle.

Adaptive Query Execution (AQE)

AQE là tính năng có từ Spark 3 và bật mặc định (spark.sql.adaptive.enabled = true từ Spark 3.2+). Ý tưởng cốt lõi: thay vì cố định kế hoạch thực thi trước khi chạy, AQE thu thập thống kê thật ngay trong lúc chạy (kích thước dữ liệu sau mỗi shuffle) rồi điều chỉnh kế hoạch cho stage tiếp theo. Nó vá đúng những điểm yếu của kế hoạch tĩnh mà ta vừa mô tả.

AQE có ba tối ưu chính:

1. Coalesce shuffle partitions động (spark.sql.adaptive.coalescePartitions.enabled = true). Bạn có thể để shuffle.partitions ở số lớn; sau shuffle, AQE nhìn kích thước thật và gộp các partition nhỏ liền kề lại cho tới khi đạt kích thước mục tiêu spark.sql.adaptive.advisoryPartitionSizeInBytes (mặc định 64MB), tôn trọng kích thước tối thiểu minPartitionSize (mặc định 1MB). Nhờ vậy bạn gần như không cần tính tay shuffle.partitions cho từng job nữa.

2. Chuyển sort-merge join → broadcast join (spark.sql.adaptive.autoBroadcastJoinThreshold). Nếu sau khi chạy một phần, AQE thấy một phía của join thực ra nhỏ hơn ngưỡng broadcast, nó chuyển từ sort-merge join (tốn shuffle cả hai phía) sang broadcast hash join (gửi bảng nhỏ tới mọi executor, không shuffle bảng lớn). Đây là điều mà kế hoạch tĩnh dựa trên ước lượng kích thước file thường bỏ lỡ. (Ngưỡng broadcast tĩnh spark.sql.autoBroadcastJoinThreshold mặc định 10MB.)

3. Xử lý skew join tự động (spark.sql.adaptive.skewJoin.enabled = true). AQE phát hiện partition skew trong sort-merge join và tách nó thành nhiều partition nhỏ hơn, chia lại thành nhiều task cân bằng. Một partition bị coi là skew khi thỏa cả hai điều kiện:

• Kích thước lớn hơn skewedPartitionFactor (mặc định 5.0) nhân với median kích thước partition, VÀ
• Kích thước vượt skewedPartitionThresholdInBytes (mặc định 256MB).

# (minh hoạ) bật & cấu hình AQE (thường đã bật sẵn từ Spark 3.2+)
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")

# tinh chỉnh nếu cần
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")

Lưu ý quan trọng: AQE skew join chỉ áp dụng cho sort-merge join, và chỉ vá được skew mà nó nhìn thấy qua thống kê shuffle. Với skew cực đoan (một key chiếm phần áp đảo) hoặc join không phải sort-merge, salting/tách key thủ công vẫn cần thiết. AQE là tuyến phòng thủ đầu tiên, không phải viên đạn bạc.

Use case thực tế: join skew làm job treo mỗi đêm

Bối cảnh. Một pipeline ETL đêm join bảng transactions (2 tỷ dòng) với accounts (5 triệu dòng) theo account_id để làm báo cáo. Job vốn chạy 20 phút, gần đây treo ở 20:00 mỗi đêm, kéo dài hơn 2 tiếng rồi thỉnh thoảng fail với ExecutorLostFailure.

Chẩn đoán. Mở Spark UI, vào stage của join:

• 599/600 task xong trong 40 giây, task cuối chạy 90+ phút → dấu hiệu straggler kinh điển.
• Summary Metrics của task: Shuffle Read Median ~90MB, nhưng Max ~48GB → chênh hàng trăm lần → skew rõ ràng.
• Cột Spill (Disk) của đúng task đó rất lớn → partition khổng lồ tràn đĩa liên tục.

Truy nguyên: một account_id là tài khoản treo trung gian (suspense account) của hệ thống thanh toán, chiếm ~70% tổng giao dịch. Hash của key đó dồn ~1.4 tỷ dòng vào một partition.

Sửa.

1. Bước một — bật/kiểm tra AQE skew join. Xác nhận spark.sql.adaptive.enabled và spark.sql.adaptive.skewJoin.enabled đều true. Sau khi bật, AQE tách partition nóng thành nhiều mảnh; thời gian tụt từ 2 tiếng xuống ~25 phút. Với đa số skew, đến đây là đủ.

2. Bước hai — salting cho key siêu nóng. Vì key này quá cực đoan, ta salt thêm để chắc chắn: bảng lớn thêm salt _0.._9 vào account_id, bảng accounts nhân bản mỗi dòng 10 lần theo salt, join theo key đã salt. Task nóng vỡ đều thành 10 phần; job ổn định ~8 phút.

3. Đo lại trên Spark UI. Max Shuffle Read của task giờ xấp xỉ Median, cột Spill về gần 0, không còn straggler. Job đêm chạy đúng SLA.

Bài học: đọc Spark UI trước, đoán sau. Median vs Max của task nói cho bạn biết đó là skew hay chỉ thiếu tài nguyên — hai vấn đề cần cách sửa khác nhau.

Ghi nhớ

• Partition = task = đơn vị song song. Quá ít → phí cluster; quá nhiều → overhead. Mục tiêu ~128–200MB/partition.
• Narrow (map/filter/select) không di chuyển dữ liệu — rẻ. Wide (groupBy/join/distinct/orderBy) cần shuffle — đắt nhất. Ranh giới shuffle = ranh giới stage.
• Shuffle = ghi đĩa → truyền mạng → đọc lại + sort. Giảm số lần shuffle và lượng dữ liệu mỗi lần (filter/select sớm).
• spark.sql.shuffle.partitions mặc định 200, cố định không theo dữ liệu → phải chỉnh, hoặc để AQE coalesce.
• Data skew = key phân bố lệch → straggler/OOM. Nhận biết qua Max ≫ Median trên Spark UI. Xử lý: salting, tách key nóng, AQE skew join.
• repartition = shuffle, cân bằng đều, tăng/giảm được. coalesce = không shuffle, chỉ giảm, có thể lệch. Ghi ít file → coalesce.
• Spill = partition không vừa RAM, tràn đĩa → chậm. Giảm bằng nhiều partition hơn + xử lý skew.
• AQE (mặc định bật từ Spark 3.2+): coalesce partition động (mục tiêu 64MB), sort-merge → broadcast, skew join tự động. Tuyến phòng thủ đầu tiên, không thay thế salting cho skew cực đoan.

Đọc tiếp

• Spark 3 — Spark SQL & DataFrame: Catalyst, plan, và nơi shuffle xuất hiện trong query.
• Spark 5 — Tối ưu & vận hành: memory, cache, broadcast join, cấu hình cluster sâu hơn.
• Spark 1 — Tổng quan kiến trúc: driver, executor, stage, task — nền tảng để hiểu bài này.