[DE Blog #06] Làm Chủ Xử Lý Dữ Liệu Phân Tán Với Apache Spark: Kiến Trúc, Shuffling & Tối Ưu Hiệu Năng PySpark

1. Bối cảnh thực tế (Context & Problem Statement)

Khi phân tích dữ liệu trên một máy tính cá nhân bằng thư viện Pandas, toàn bộ dữ liệu bắt buộc phải nạp vừa vào bộ nhớ RAM đơn lẻ. Khi tập dữ liệu tăng từ vài Gigabyte lên hàng trăm Gigabyte hoặc Terabyte, lệnh pd.read_csv() sẽ ngay lập tức gây sập chương trình với lỗi kinh điển: MemoryError (Out of Memory - OOM).

Apache Spark ra đời để giải quyết bài toán này bằng mô hình Tính toán phân tán trên bộ nhớ (In-Memory Distributed Computing): Thay vì cố gắng mua một máy chủ khổng lồ (Vertical Scaling), Spark chia nhỏ dữ liệu thành nhiều phân vùng (Partitions) và phân phối cho hàng chục hoặc hàng trăm máy chủ nhỏ (Worker Nodes) tính toán song song.

2. Các Khái Niệm & Cơ Chế Cốt Lõi

2.1. Kiến Trúc Phân Tán Của Apache Spark

Hệ thống Spark hoạt động theo mô hình Master-Worker:

Driver Program (Master): Nơi chạy hàm main(), khởi tạo SparkSession, chuyển đổi code của người dùng thành đồ thị thực thi DAG (Directed Acyclic Graph), sau đó lập lịch và phân phối các Task tới các Executor.
Cluster Manager: Quản lý và phân bổ tài nguyên phần cứng (YARN, Kubernetes, hoặc Standalone).
Executors (Workers): Các tiến trình chạy trên Worker Nodes, nhận Task từ Driver, trực tiếp thực thi tính toán và lưu trữ dữ liệu (RAM + Disk).

+-------------------------------------------------------------+
|                        Driver Program                       |
|          (SparkSession, DAGScheduler, TaskScheduler)        |
+------------------------------+------------------------------+
                               |
                               v
                     +-------------------+
                     |  Cluster Manager  | (K8s / YARN)
                     +---------+---------+
                               |
            +------------------+------------------+
            |                                     |
            v                                     v
    +---------------+                     +---------------+
    |   Executor 1  |                     |   Executor 2  |
    | [Task][Task]  |                     | [Task][Task]  |
    | (BlockManager)|                     | (BlockManager)|
    +---------------+                     +---------------+

2.2. Lazy Evaluation & Vòng Đời Thực Thi (Transformations vs. Actions)

Lazy Evaluation (Đánh giá lười biếng): Spark không thực thi tính toán ngay khi bạn gọi các hàm biến đổi dữ liệu. Thay vào đó, nó chỉ ghi nhận logic vào một đồ thị kế hoạch thực thi (Lineage Graph / DAG).
Transformations (Tạo DataFrame mới từ DataFrame cũ):
- Narrow Transformations: Mỗi partition của DataFrame cha chỉ phụ thuộc vào tối đa 1 partition của DataFrame con (ví dụ: map, filter, select). Không phát sinh di chuyển dữ liệu qua mạng.
- Wide Transformations: Dữ liệu từ nhiều partition cần được gom nhóm và phân phối lại trên toàn bộ cluster (ví dụ: groupBy, join, distinct, orderBy). Gây ra hiện tượng Shuffle.
Actions (Kích hoạt thực thi thực tế): Chỉ khi người dùng gọi một Action (ví dụ: count(), collect(), show(), write.parquet()), Spark mới tổng hợp toàn bộ DAG, tối ưu hóa qua Catalyst Optimizer và bắt đầu gửi Task xuống các Executor.

2.3. Data Shuffling & Hiện Tượng Data Skew

Data Shuffling (Xáo trộn dữ liệu): Quá trình tái phân phối dữ liệu qua mạng giữa các Executor để phục vụ các phép tính Wide Transformation. Đây là nguyên nhân số 1 gây chậm job và nghẽn I/O.
Data Skew (Lệch dữ liệu): Hiện tượng dữ liệu phân bổ không đồng đều giữa các partition (ví dụ: 1 key chiếm 90% số dòng). Kết quả: 99% các task chạy xong trong vài giây, nhưng 1 task cuối cùng bị treo hàng tiếng đồng hồ hoặc chết vì OOM.

3. Các Bảng Markdown So Sánh Chi Tiết

📊 Bảng 1: So Sánh RDD vs. DataFrame vs. Dataset

Tiêu chí	RDD (Resilient Distributed Dataset)	DataFrame	Dataset
Bản chất	Tập hợp các đối tượng Java/Python phân tán nguyên thủy	Dữ liệu dạng bảng có Schema rõ ràng (các cột có tên và kiểu)	Kết hợp tính an toàn kiểu dữ liệu của RDD + Tối ưu của DataFrame
Ngôn ngữ hỗ trợ	Scala, Java, Python, R	Scala, Java, Python, R	Chỉ hỗ trợ Scala và Java (Không có trong PySpark)
Công cụ tối ưu hóa	Không có (Lập trình viên tự chịu trách nhiệm tối ưu)	Catalyst Optimizer & Tungsten Engine tối ưu tự động	Tận dụng Catalyst Optimizer và Encoder chuyên dụng
Kiểm tra kiểu (Type-safety)	Compile-time (với Scala/Java)	Run-time (dễ lỗi khi sai tên cột)	Compile-time (bắt lỗi kiểu dữ liệu ngay khi biên dịch)
Khuyến nghị sử dụng	Chỉ dùng khi cần can thiệp cấp thấp vào partition	Chuẩn mực mặc định cho hầu hết các tác vụ Data Engineering	Khi làm việc với ứng dụng Scala/Java yêu cầu type-safety cao

📊 Bảng 2: So Sánh Narrow Transformation vs. Wide Transformation

Tiêu chí	Narrow Transformation	Wide Transformation
Quan hệ giữa các Partition	$1 \rightarrow 1$ (hoặc Nhiều $\rightarrow 1$ không phụ thuộc chéo)	Nhiều $\rightarrow$ Nhiều (Phụ thuộc chéo phức tạp)
Di chuyển dữ liệu qua mạng	Không có (Zero Network I/O)	Bắt buộc phát sinh Shuffle qua mạng
Các hàm tiêu biểu	`filter()`, `select()`, `map()`, `union()`, `drop()`	`groupBy()`, `join()`, `distinct()`, `repartition()`, `reduceByKey()`
Tốc độ thực thi	Cực nhanh, tính toán độc lập tại từng Node	Chậm hơn, phụ thuộc vào băng thông mạng và đĩa đệm (Disk Spill)

📊 Bảng 3: So Sánh Các Chiến Lược JOIN Trong Spark

Thuật toán Join	Điều kiện kích hoạt tối ưu	Cơ chế hoạt động	Chi phí Shuffle
Broadcast Hash Join (BHJ)	Một trong hai bảng rất nhỏ (mặc định $\le 10\text{ MB}$ , có thể cấu hình lên hàng trăm MB)	Driver tải toàn bộ bảng nhỏ về và gửi (broadcast) đến tất cả Executor	Không có Shuffle (Zero Shuffle)
Shuffle Hash Join	2 bảng lớn nhưng 1 bảng nhỏ hơn đáng kể so với bảng còn lại, bảng nhỏ đủ chứa trong bộ nhớ	Cả 2 bảng đều bị shuffle theo Join Key, bảng nhỏ hơn được dùng để build Hash Table trên từng Executor	Có Shuffle toàn bộ
Sort-Merge Join (SMJ)	2 bảng đều rất lớn (Default của Spark khi join 2 bảng lớn)	Cả 2 bảng bị Shuffle theo Join Key, sau đó Sort trên từng partition rồi Merge lại	Có Shuffle và chi phí Sort cao

4. Best Practices & Kỹ Thuật Tối Ưu PySpark (Pro-Tips & Tuning)

💡 Chủ động sử dụng Broadcast Join cho bảng kích thước nhỏ

Khi join bảng Fact (hàng trăm triệu dòng) với bảng Dimension (vài chục nghìn dòng):

from pyspark.sql.functions import broadcast

# Ép Spark sử dụng Broadcast Join để triệt tiêu hoàn toàn Shuffle:
fct_sales_df = spark.read.parquet("s3://data-lake/fct_sales")
dim_store_df = spark.read.parquet("s3://data-lake/dim_store")

result_df = fct_sales_df.join(
    broadcast(dim_store_df), 
    on="store_id", 
    how="inner"
)

💡 Phân biệt rõ `repartition()` và `coalesce()`

coalesce(n): Chỉ dùng để giảm số lượng partition. Nó gộp các partition liền kề trên cùng node $\rightarrow$ Không gây ra Full Shuffle (rất tối ưu khi chuẩn bị ghi file ra đĩa để tránh Small Files Problem).
repartition(n, [cols]): Dùng để tăng hoặc giảm số lượng partition, phân bổ lại dữ liệu đồng đều dựa trên Hash $\rightarrow$ Luôn gây ra Full Shuffle.

💡 Kỹ thuật "Salting" để giải quyết triệt để Data Skew

Khi một giá trị khóa (ví dụ user_id = NULL hoặc store_id = 9999) chiếm đa số dữ liệu, phép JOIN hoặc groupBy sẽ bị nghẽn:

Bước 1 (Bảng Skew): Thêm một cột số nguyên ngẫu nhiên từ $0$ đến $K-1$ vào khóa chính: salted_key = concat(key, "_", floor(rand() * K)).
Bước 2 (Bảng Lookup): Nhân bản (Replicate) mỗi dòng của bảng lookup thành $K$ dòng tương ứng với các giá trị từ $0$ đến $K-1$ .
Bước 3: Thực hiện phép JOIN trên salted_key. Dữ liệu skew sẽ được chia đều sang $K$ worker khác nhau để tính toán song song.

💡 Tuyệt đối tránh gọi `df.collect()` trên tập dữ liệu lớn

Lệnh collect() sẽ kéo toàn bộ dữ liệu từ tất cả các Executor về duy nhất bộ nhớ của Driver Node. Nếu dữ liệu lớn hơn RAM của Driver, ứng dụng sẽ sập ngay lập tức. Hãy dùng take(n), limit(n) hoặc lưu thẳng kết quả ra Data Lake (write.parquet).

5. Góc Ôn Luyện Phỏng Vấn (Interview Corner)

❓ Câu hỏi 1: Giải thích cơ chế Lazy Evaluation trong Apache Spark. Nó giúp tối ưu hóa hiệu năng như thế nào qua Catalyst Optimizer?

Gợi ý trả lời:
- Định nghĩa: Lazy Evaluation nghĩa là các phép Transformation không tính toán ngay mà chỉ xây dựng đồ thị quan hệ logic (DAG Lineage). Việc tính toán thực tế chỉ bắt đầu khi một Action được gọi.
- Lợi ích tối ưu hóa của Catalyst Optimizer:
  1. Predicate Pushdown: Nếu câu lệnh có filter() ở cuối chuỗi biến đổi, Catalyst sẽ tự động đẩy bộ lọc này xuống trước các phép toán nặng hoặc đẩy thẳng xuống tầng đọc file (Parquet) để giảm lượng dữ liệu cần nạp vào RAM.
  2. Column Pruning: Tự động loại bỏ các cột không được sử dụng trong suốt pipeline trước khi thực hiện các phép Join/Shuffle.
  3. Query Reordering: Tối ưu hóa thứ tự các phép toán để giảm chi phí trung gian.

❓ Câu hỏi 2: Data Shuffling trong Spark là gì? Tại sao nó lại là kẻ thù số 1 của hiệu năng và làm thế nào để giảm thiểu Shuffle?

Gợi ý trả lời:
- Bản chất: Shuffle là quá trình phân phối lại dữ liệu trên toàn bộ cụm máy chủ khi thực hiện các Wide Transformation (groupBy, join, distinct). Nó bao gồm 3 bước tốn kém: Ghi dữ liệu trung gian ra đĩa (Disk I/O) $\rightarrow$ Truyền tải dữ liệu qua mạng giữa các node (Network I/O) $\rightarrow$ Đọc và sắp xếp lại dữ liệu trên node nhận.
- Cách hạn chế/tối ưu Shuffle:
  - Sử dụng Broadcast Join khi một trong các bảng có kích thước nhỏ.
  - Sử dụng reduceByKey thay cho groupByKey (với RDD) để tận dụng Map-side Combine trước khi gửi qua mạng.
  - Sử dụng coalesce() thay vì repartition() khi chỉ cần giảm số lượng file đầu ra.
  - Lọc dữ liệu (filter) và loại bỏ cột thừa (select) càng sớm càng tốt trước khi gọi phép Join.

❓ Câu hỏi 3: Khi chạy một Spark Job, bạn nhận thấy job bị "treo" ở task cuối cùng (99%) trong suốt 45 phút rồi báo lỗi OutOfMemoryError. Đây là dấu hiệu của vấn đề gì và cách xử lý của bạn là gì?

Gợi ý trả lời:
- Chẩn đoán: Đây là triệu chứng kinh điển của Data Skew (Lệch dữ liệu). Phép toán Wide Transformation phân bổ dữ liệu không đều, khiến một partition chứa lượng dữ liệu khổng lồ rơi vào một Executor duy nhất, trong khi các Executor khác đã chạy xong.
- Các bước xử lý:
  1. Mở Spark UI, vào tab Stages $\rightarrow$ Xem biểu đồ Task Deserialization / Shuffle Read Size. Nếu có sự chênh lệch lớn giữa Max và Median/75th percentile, chắc chắn là Data Skew.
  2. Kiểm tra dữ liệu nguồn xem có giá trị NULL hoặc các giá trị mặc định chiếm tỷ trọng áp đảo không. Nếu có, tách riêng tập dữ liệu đó để xử lý riêng hoặc lọc trước khi join.
  3. Áp dụng kỹ thuật Salting (băm khóa với số ngẫu nhiên) để phân tán khóa bị skew sang nhiều partition.
  4. Bật tính năng Adaptive Query Execution (AQE) của Spark 3.0+ (spark.sql.adaptive.skewJoin.enabled = true) để Spark tự động chia nhỏ các partition bị skew trong lúc chạy.

6. Tóm Tắt Ghi Nhớ Nhanh (Key Takeaways)

Spark tối ưu nhờ In-Memory Computing và Lazy Evaluation thông qua Catalyst Optimizer.
Narrow Transformation không phát sinh Shuffle; Wide Transformation luôn phát sinh Shuffle qua mạng.
Luôn ưu tiên Broadcast Join khi join với bảng nhỏ để triệt tiêu chi phí Shuffle.
Data Skew là nguyên nhân hàng đầu khiến task bị treo ở 99% $\rightarrow$ Giải quyết bằng AQE hoặc kỹ thuật Salting.