Data Engineering 4 — Apache Spark & Xử lý phân tán

Vì sao một máy không đủ?

Trong ba bài trước, bạn đã quen với việc xử lý dữ liệu trên một máy duy nhất: đọc một file CSV, chạy một câu SQL, transform vài chục nghìn dòng. Cách làm này hoạt động tốt cho đến khi dữ liệu lớn đến mức không một chiếc máy nào đủ sức xử lý trong thời gian chấp nhận được. Đây là lúc khái niệm "big data" trở nên cụ thể.

Người ta thường mô tả big data bằng "3 chữ V":

• Volume (khối lượng): Dữ liệu lên tới hàng terabyte hoặc petabyte. Một file 10 TB không thể nạp vào RAM của một máy (thường chỉ vài chục đến vài trăm GB). Thậm chí đọc tuần tự từ ổ đĩa cũng mất hàng giờ.
• Velocity (tốc độ): Dữ liệu sinh ra liên tục với tốc độ rất cao — log của hệ thống ngân hàng, sự kiện click trên web, dữ liệu cảm biến. Một máy đơn không kịp tiêu thụ.
• Variety (đa dạng): Dữ liệu đến từ nhiều nguồn và định dạng khác nhau — JSON, Parquet, ảnh, văn bản tự do, bản ghi quan hệ.

Cốt lõi của vấn đề là giới hạn vật lý. Một CPU có số nhân hữu hạn, RAM có dung lượng hữu hạn, ổ đĩa có băng thông đọc/ghi hữu hạn. Bạn có thể nâng cấp máy mạnh hơn (gọi là scale up / mở rộng theo chiều dọc), nhưng chi phí tăng theo cấp số nhân và luôn có một trần cứng: không tồn tại chiếc máy nào có 10.000 nhân CPU và 100 TB RAM với giá hợp lý.

Giải pháp thay thế là scale out (mở rộng theo chiều ngang): thay vì một máy khổng lồ, dùng hàng trăm máy bình thường (commodity hardware) nối với nhau thành một cluster, rồi chia công việc cho chúng cùng làm song song. Đây chính là tư tưởng của xử lý phân tán (distributed processing).

Xử lý phân tán & MapReduce

Khi chia việc cho nhiều máy, ta cần một mô hình lập trình để diễn đạt "chia nhỏ — xử lý song song — gộp lại" một cách có hệ thống. Mô hình kinh điển nhất là MapReduce, do Google đề xuất năm 2004.

Ý tưởng của MapReduce là tư duy chia-để-trị (divide and conquer) được chuẩn hóa thành ba pha:

Map

Mỗi máy nhận một phần dữ liệu (một partition) và áp dụng một hàm map lên từng bản ghi, sinh ra các cặp (key, value). Pha này hoàn toàn song song — các máy không cần nói chuyện với nhau.

Ví dụ kinh điển: đếm số lần xuất hiện của mỗi từ trong một kho văn bản khổng lồ. Hàm map đọc một dòng và phát ra (từ, 1) cho mỗi từ.

Shuffle

Đây là pha gom nhóm theo key: tất cả các cặp có cùng key được đưa về cùng một máy. Để làm vậy, dữ liệu phải di chuyển qua mạng giữa các máy. Pha shuffle là phần tốn kém nhất, ta sẽ quay lại sau.

Reduce

Mỗi máy nhận toàn bộ value của một (hoặc một nhóm) key và áp dụng hàm reduce để tổng hợp. Trong ví dụ đếm từ, reduce cộng tất cả các số 1 của cùng một từ lại để ra tổng số lần xuất hiện.

MapReduce mạnh ở chỗ lập trình viên chỉ cần viết hai hàm map và reduce; phần phức tạp — phân phối dữ liệu, song song hóa, xử lý máy hỏng — do framework lo. Nhược điểm là mỗi job phải ghi kết quả trung gian xuống đĩa rồi đọc lại cho job sau, khiến các pipeline nhiều bước (như machine learning lặp đi lặp lại) chạy rất chậm.

Hadoop/HDFS vs Spark

Hadoop là hệ sinh thái mã nguồn mở hiện thực hóa MapReduce, gồm hai phần chính:

• HDFS (Hadoop Distributed File System): hệ thống file phân tán. Một file lớn được chia thành các block (mặc định 128 MB) và sao chép (replicate, thường 3 bản) trên nhiều máy. Nhờ vậy dữ liệu vừa được lưu phân tán vừa chịu được lỗi (một máy chết vẫn còn bản sao).
• MapReduce engine: chạy các job tính toán trên dữ liệu nằm trong HDFS, ưu tiên đưa tính toán đến gần dữ liệu (data locality) thay vì kéo dữ liệu về.

Hadoop MapReduce rất bền bỉ nhưng chậm vì ghi/đọc đĩa giữa các bước. Apache Spark ra đời để giải quyết đúng điểm yếu này.

Khác biệt cốt lõi: Spark giữ dữ liệu trung gian trong bộ nhớ (in-memory) giữa các bước thay vì ghi xuống đĩa. Với các workload lặp lại, Spark có thể nhanh hơn Hadoop MapReduce hàng chục đến cả trăm lần. Spark cũng cung cấp API cấp cao (SQL, DataFrame, streaming, ML) thay vì bắt bạn viết tay từng cặp map/reduce. Lưu ý: Spark là một engine tính toán — nó không thay thế HDFS mà thường đọc dữ liệu từ HDFS, S3, hoặc các nguồn khác.

Kiến trúc Spark

Một ứng dụng Spark chạy theo mô hình master–worker với ba thành phần chính:

Driver

Là "bộ não" của ứng dụng. Driver chạy chương trình chính của bạn (main), tạo ra SparkSession, phân tích code để dựng kế hoạch thực thi, chia công việc thành các task, và điều phối/lập lịch các task đó. Driver cũng là nơi nhận kết quả cuối cùng khi bạn gọi collect().

Executor

Là các tiến trình chạy trên các máy worker, thực thi các task mà driver giao. Mỗi executor có một số nhân CPU và một vùng bộ nhớ riêng để chạy task và cache dữ liệu. Đây là nơi công việc thực sự diễn ra, song song trên nhiều máy.

Cluster Manager

Là thành phần cấp phát tài nguyên (CPU, RAM) cho ứng dụng — quyết định khởi chạy executor ở đâu, bao nhiêu. Spark hỗ trợ nhiều cluster manager: standalone, YARN (của Hadoop), Mesos, hoặc Kubernetes.

RDD vs DataFrame/Dataset

Spark cung cấp nhiều mức trừu tượng để biểu diễn dữ liệu phân tán.

RDD (Resilient Distributed Dataset)

Là khái niệm nền tảng đầu tiên: một tập hợp bất biến (immutable), phân tán các phần tử, được chia thành nhiều partition trên cluster. "Resilient" nghĩa là chịu lỗi — Spark ghi lại lineage (chuỗi các phép biến đổi tạo ra RDD), nên nếu một partition mất, nó có thể được tính lại từ đầu.

RDD cho bạn quyền kiểm soát ở mức thấp, nhưng Spark không hiểu cấu trúc bên trong dữ liệu nên không tối ưu được nhiều. Bạn cũng phải tự viết logic xử lý.

DataFrame

Là một tập dữ liệu phân tán được tổ chức thành các cột có tên — giống một bảng trong cơ sở dữ liệu hay một DataFrame của pandas. Vì Spark biết schema (tên cột, kiểu dữ liệu), nó có thể dùng bộ tối ưu Catalyst để viết lại và tối ưu kế hoạch truy vấn tự động. Đây là API được khuyên dùng nhất, đặc biệt trong PySpark.

Dataset

Là phiên bản có kiểm tra kiểu (type-safe) của DataFrame, chủ yếu dùng trong Scala/Java. Trong PySpark (Python), bạn làm việc gần như hoàn toàn với DataFrame.

Quy tắc thực dụng: dùng DataFrame/Spark SQL trong hầu hết trường hợp; chỉ rơi xuống RDD khi cần kiểm soát rất chi tiết mà API cấp cao không đáp ứng được.

Transformation (lazy) vs Action

Các thao tác trên DataFrame/RDD chia làm hai loại, và phân biệt được chúng là chìa khóa để hiểu Spark.

• Transformation: tạo ra một dataset mới từ dataset cũ, ví dụ filter(), select(), map(), groupBy(), join(). Transformation lazy (lười) — gọi nó không hề chạy tính toán, chỉ ghi lại "ý định".
• Action: kích hoạt tính toán thực sự và trả về kết quả, ví dụ count(), collect(), show(), write(). Chỉ khi gặp action, Spark mới thực sự chạy chuỗi transformation đã tích lũy.

Lazy Evaluation & DAG thực thi

Vì sao Spark lại lười? Bởi vì hoãn tính toán cho phép tối ưu. Khi bạn xâu chuỗi nhiều transformation, Spark không chạy ngay từng cái mà dựng một DAG (Directed Acyclic Graph) — đồ thị có hướng không chu trình — mô tả toàn bộ luồng biến đổi.

Khi action được gọi, Spark nhìn vào toàn bộ DAG và tối ưu trước khi chạy: ví dụ đẩy phép filter lên sớm nhất có thể (predicate pushdown) để giảm dữ liệu cần xử lý, hoặc gộp nhiều phép vào một bước. Sau đó nó chia DAG thành các stage, mỗi stage gồm nhiều task chạy song song trên các partition.

Hệ quả thực tế: nếu bạn viết một loạt transformation rồi không bao giờ gọi action, không có gì chạy cả. Đây là điều khiến người mới hay bối rối khi thấy code "không làm gì".

Partition & Shuffle

Partition là đơn vị song song của Spark: một DataFrame được chia thành nhiều partition, mỗi partition do một task xử lý trên một nhân CPU. Số partition quyết định mức độ song song.

Shuffle là quá trình phân phối lại dữ liệu giữa các partition (và thường là giữa các máy qua mạng). Nó xảy ra khi một thao tác cần gom các bản ghi có cùng key về cùng chỗ — ví dụ groupBy, join, distinct, repartition.

Vì sao shuffle tốn kém? Vì nó kéo theo:

1. Ghi đĩa: dữ liệu shuffle được ghi ra đĩa cục bộ ở phía gửi.
2. Truyền mạng: dữ liệu phải đi qua mạng giữa các executor — chậm hơn RAM nhiều bậc.
3. Serialize/deserialize: dữ liệu phải được đóng gói rồi giải gói.
4. Rào chắn (barrier): stage sau phải chờ shuffle của stage trước hoàn tất.

Nguyên tắc tối ưu Spark số một là giảm thiểu shuffle.

Narrow vs Wide Dependency

Cách Spark phân chia stage dựa trên loại phụ thuộc giữa các partition:

• Narrow dependency (phụ thuộc hẹp): mỗi partition đầu vào chỉ đóng góp cho một partition đầu ra. Ví dụ map, filter. Không cần di chuyển dữ liệu qua mạng — chạy được hoàn toàn cục bộ, rất nhanh.
• Wide dependency (phụ thuộc rộng): một partition đầu vào có thể đóng góp cho nhiều partition đầu ra, đòi hỏi gom dữ liệu từ nhiều nguồn. Ví dụ groupByKey, join. Đây chính là cái sinh ra shuffle.

Mỗi khi gặp wide dependency, Spark cắt một stage boundary (ranh giới stage). Vì vậy nhìn vào DAG và đếm số shuffle, bạn ước lượng được chi phí của job.

Spark SQL & PySpark

PySpark là API Python của Spark. Dưới đây là một ví dụ ngắn minh họa DataFrame, transformation lazy, và action.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# Tạo điểm vào của ứng dụng Spark
spark = SparkSession.builder.appName("vi-du").getOrCreate()

# Đọc dữ liệu (Parquet được khuyến nghị: có schema, nén tốt)
don_hang = spark.read.parquet("s3://bucket/don_hang/")

# Các transformation dưới đây đều LAZY - chưa chạy gì
ket_qua = (
    don_hang
    .filter(F.col("trang_thai") == "thanh_cong")   # narrow
    .groupBy("ma_khach_hang")                       # wide -> shuffle
    .agg(F.sum("gia_tri").alias("tong_chi_tieu"))
    .orderBy(F.desc("tong_chi_tieu"))
)

# ACTION -> bây giờ Spark mới dựng DAG, tối ưu và chạy
ket_qua.show(10)

Bạn cũng có thể dùng SQL thuần trên dữ liệu phân tán:

don_hang.createOrReplaceTempView("don_hang")

spark.sql("""
    SELECT ma_khach_hang, SUM(gia_tri) AS tong_chi_tieu
    FROM don_hang
    WHERE trang_thai = 'thanh_cong'
    GROUP BY ma_khach_hang
    ORDER BY tong_chi_tieu DESC
""").show(10)

Cả hai cách đều đi qua cùng một bộ tối ưu Catalyst nên hiệu năng tương đương — chọn cách nào dễ đọc hơn với nhóm của bạn.

Tối ưu Spark

Partitioning hợp lý

Quá ít partition thì không tận dụng hết cluster; quá nhiều partition thì chi phí lập lịch và overhead lấn át. Một quy tắc thường gặp là nhắm tới partition cỡ vài trăm MB. Dùng repartition(n) (gây shuffle) khi cần tăng song song, hoặc coalesce(n) (không shuffle, chỉ gộp) khi cần giảm số partition đầu ra.

Broadcast join

Khi join một bảng lớn với một bảng nhỏ, thay vì shuffle cả hai, Spark có thể gửi (broadcast) toàn bộ bảng nhỏ tới mọi executor để join cục bộ — loại bỏ shuffle hoàn toàn.

from pyspark.sql.functions import broadcast

ket_qua = don_hang.join(broadcast(danh_muc_san_pham), "ma_san_pham")

Caching

Nếu một DataFrame được dùng lại nhiều lần (ví dụ trong vòng lặp ML), hãy cache() hoặc persist() để Spark giữ nó trong bộ nhớ thay vì tính lại từ đầu mỗi lần. Đừng cache bừa bãi: bộ nhớ có hạn.

Tránh data skew (lệch dữ liệu)

Skew xảy ra khi một vài key có quá nhiều bản ghi (ví dụ giá trị null hoặc một khách hàng VIP), khiến một task xử lý lượng dữ liệu khổng lồ trong khi các task khác đã xong từ lâu. Cả job phải chờ task chậm nhất. Cách xử lý: lọc/tách riêng key lệch, hoặc dùng kỹ thuật salting (thêm hậu tố ngẫu nhiên vào key để rải đều), hoặc bật Adaptive Query Execution (AQE) của Spark để nó tự phát hiện và chia nhỏ partition lệch.

Khi nào KHÔNG cần Spark?

Spark mạnh nhưng đi kèm chi phí: phức tạp khi vận hành cluster, overhead khởi động JVM/executor, và việc debug job phân tán khó hơn nhiều so với một script đơn lẻ.

Bạn không cần Spark khi:

• Dữ liệu vừa với một máy. Nếu dataset vài GB chạy được trên pandas, DuckDB, hoặc Polars trên một máy thì các công cụ này nhanh hơn và đơn giản hơn nhiều — không có chi phí mạng, không có shuffle.
• Khối lượng nhỏ nhưng tần suất cao. Với các truy vấn nhỏ liên tục, một cơ sở dữ liệu OLTP/OLAP thông thường phù hợp hơn.
• Bài toán đã có công cụ chuyên dụng tốt hơn, ví dụ một data warehouse (BigQuery, Snowflake) đã đủ cho phân tích SQL.

Nguyên tắc vàng: đừng dùng cluster để giải bài toán của một máy. Hãy đo dữ liệu thực tế trước, và chỉ chọn Spark khi quy mô thực sự vượt khả năng của một máy đơn.

Tóm tắt

• Một máy có giới hạn vật lý (RAM/CPU/đĩa); với big data (volume/velocity/variety) ta cần scale out — chia việc cho nhiều máy.
• MapReduce chuẩn hóa tư duy chia-để-trị thành ba pha map → shuffle → reduce. Hadoop/HDFS hiện thực hóa nó nhưng chậm vì ghi đĩa; Spark nhanh hơn nhiều nhờ xử lý in-memory.
• Kiến trúc Spark gồm driver (điều phối), executor (thực thi), cluster manager (cấp tài nguyên).
• DataFrame/Spark SQL là API nên dùng (Catalyst tối ưu tự động); RDD chỉ khi cần kiểm soát thấp.
• Transformation lazy, chỉ chạy khi gặp action; Spark dựng và tối ưu DAG trước khi thực thi.
• Shuffle (sinh bởi wide dependency) là phần tốn kém nhất — giảm shuffle là ưu tiên tối ưu hàng đầu, cùng với broadcast join, caching và xử lý skew.
• Đừng dùng Spark nếu dữ liệu vừa với một máy.

Tự kiểm tra

1. Giải thích sự khác nhau giữa scale up và scale out, và vì sao big data thường buộc ta chọn scale out.
2. Pha shuffle trong MapReduce/Spark làm gì, và tại sao nó tốn kém hơn nhiều so với pha map?
3. Phân biệt transformation và action. Điều gì xảy ra nếu một chuỗi transformation không bao giờ được theo sau bởi một action?
4. Narrow dependency và wide dependency khác nhau ra sao? Loại nào tạo ra ranh giới stage và shuffle?
5. Khi join một bảng lớn với một bảng nhỏ, kỹ thuật nào giúp tránh shuffle, và nó hoạt động thế nào?
6. Nêu ít nhất hai tình huống mà bạn KHÔNG nên dùng Spark, kèm lý do.

Đọc tiếp

Data Engineering 5 — Kafka & Streaming