Spark 2 — RDD, DataFrame, Dataset & Catalyst

Bài Tổng quan Spark trả lời câu hỏi "Spark là gì và chạy trên cụm ra sao". Bài này đi sâu vào thứ bạn thực sự gõ mỗi ngày: API dữ liệu. Spark cung cấp không phải một, mà ba tầng API để biểu diễn và xử lý dữ liệu phân tán — RDD, DataFrame và Dataset. Ba tầng này khác nhau ở mức độ trừu tượng, ở việc có schema hay không, và quan trọng nhất: ở chỗ bạn phải tự tối ưu hay để Spark tối ưu giùm.

Hiểu rõ khác biệt giữa chúng — cùng với cơ chế Catalyst và Tungsten đứng sau — là điều phân biệt người "viết được code Spark" với người "viết được code Spark nhanh". Toàn bộ bài dựa trên Apache Spark 3.5+.

Lưu ý môi trường: sandbox của knowledge base này chạy PostgreSQL, nên PySpark không thực thi được ở đây. Mọi khối python/scala dưới đây là (minh hoạ) để đọc và hiểu tư duy, không phải để copy-paste chạy trực tiếp.

Bức tranh ba tầng API

Hãy hình dung ba tầng như ba mức độ "nói chuyện" với Spark:

Điểm mấu chốt cần nắm ngay từ đầu: RDD là API "điều khiển thấp" (low-level) — bạn ra lệnh làm thế nào; còn DataFrame/Dataset là API "khai báo" (declarative) — bạn mô tả muốn gì, để Spark tự quyết định làm thế nào. Chính sự khác biệt "mô tả cái gì" thay vì "chỉ đạo từng bước" là điều mở đường cho tối ưu tự động.

RDD — nền móng của Spark

RDD viết tắt của Resilient Distributed Dataset — tập dữ liệu phân tán, có khả năng phục hồi. Đây là kiểu trừu tượng đầu tiên và nền tảng nhất của Spark. Ba đặc tính trong cái tên nói lên tất cả:

• Resilient (phục hồi được): nếu một partition mất do node hỏng, Spark tính lại nó từ lineage — chuỗi phép biến đổi đã tạo ra nó — chứ không cần sao chép dữ liệu thô. Đây là cơ chế chịu lỗi cốt lõi của Spark.
• Distributed (phân tán): dữ liệu được chia thành nhiều partition rải trên các executor, xử lý song song.
• Dataset (tập dữ liệu): một tập hợp các phần tử — có thể là số, chuỗi, tuple, hay object Python/Scala tùy ý.

RDD bất biến (immutable): mọi phép biến đổi tạo ra một RDD mới chứ không sửa RDD cũ. API của RDD mang phong cách lập trình hàm — bạn xâu chuỗi các phép map, filter, flatMap, reduce, reduceByKey...

# (minh hoạ) — Đếm từ bằng RDD thuần
rdd = sc.textFile("hdfs:///logs/access.log")
counts = (
    rdd.flatMap(lambda line: line.split(" "))   # tách từ
       .map(lambda word: (word, 1))              # (từ, 1)
       .reduceByKey(lambda a, b: a + b)          # cộng dồn theo từ
)
counts.saveAsTextFile("hdfs:///out/wordcount")

Đặc điểm sống còn cần hiểu: RDD KHÔNG có schema và KHÔNG được tối ưu tự động. Với Spark, một RDD chỉ là "một đống object mờ đục" (opaque). Nó không biết bên trong mỗi phần tử có cột user_id hay amount, cũng không biết kiểu dữ liệu. Vì thế Spark không thể áp các tối ưu như bỏ cột thừa hay đẩy điều kiện lọc xuống nguồn — nó buộc phải chạy đúng từng lệnh bạn viết, theo thứ tự bạn viết. Hàm lambda bạn truyền vào là hộp đen với Spark.

Khi nào vẫn cần RDD?

Trong công việc hằng ngày với PySpark, bạn nên mặc định dùng DataFrame. Nhưng RDD vẫn có chỗ đứng:

• Cần điều khiển chi tiết cách phân vùng dữ liệu (custom partitioner) hoặc thao tác ở mức partition (mapPartitions).
• Dữ liệu không có cấu trúc rõ ràng — văn bản thô, media, dữ liệu nhị phân — mà mô hình bảng không phù hợp.
• Xử lý logic quá tùy biến, không diễn đạt được bằng hàm dựng sẵn hay SQL.
• Làm việc với thư viện/RDD cũ, hoặc cần truy cập lineage/checkpoint ở mức thấp.

Ngoài các trường hợp trên, viết bằng RDD thuần thường chậm hơn DataFrame làm cùng việc — nghe có vẻ ngược đời, nhưng phần Catalyst dưới đây sẽ giải thích vì sao.

DataFrame & Dataset — dữ liệu có cấu trúc

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 và kiểu — về mặt khái niệm giống một bảng trong cơ sở dữ liệu quan hệ, hay một data.frame/pandas DataFrame nhưng phân tán và lười tính toán.

Với DataFrame, bạn thao tác bằng API khai báo: chọn cột, lọc, nhóm, join — bằng biểu thức chứ bằng lambda hộp đen.

# (minh hoạ) — Cùng bài word count nhưng bằng DataFrame API
from pyspark.sql import functions as F

df = spark.read.text("hdfs:///logs/access.log")
counts = (
    df.select(F.explode(F.split(F.col("value"), " ")).alias("word"))
      .groupBy("word")
      .count()
)
counts.write.parquet("hdfs:///out/wordcount")

Điểm khác biệt tinh tế nhưng cực quan trọng: ở phiên bản DataFrame, Spark hiểu bạn đang tách chuỗi, gom nhóm, đếm. Nó nhìn thấy cấu trúc của phép tính, nên có thể sắp xếp lại và tối ưu.

Dataset và quan hệ với DataFrame

Dataset là API có kiểu tĩnh (typed), chỉ có trong Scala và Java — nơi trình biên dịch kiểm tra kiểu lúc build. Điều cần nhớ gọn:

• Dataset[T] cho phép làm việc với object có kiểu cụ thể T (ví dụ Dataset[User]), được kiểm tra kiểu lúc biên dịch.
• DataFrame chính là Dataset[Row] — tức Dataset của các dòng không định kiểu. Trong Scala, DataFrame chỉ là bí danh của Dataset[Row].

// (minh hoạ, Scala) — Dataset typed
case class User(id: Long, country: String, amount: Double)

val ds: Dataset[User] = spark.read.parquet("s3://.../users").as[User]
val vn = ds.filter(_.country == "VN")   // kiểm tra kiểu lúc biên dịch

PySpark không có Dataset typed — Python là ngôn ngữ động, không có kiểm tra kiểu lúc biên dịch. Vì vậy trong PySpark, bạn chủ yếu dùng DataFrame, và đó cũng là API bạn sẽ gặp 90% thời gian. Điều đáng mừng: cả DataFrame và Dataset đều đi qua cùng một bộ tối ưu, nên về hiệu năng chúng tương đương nhau.

Catalyst — vì sao DataFrame nhanh hơn RDD viết tay

Đây là phần "ma thuật". Khi bạn viết DataFrame/Dataset/SQL, Spark không chạy ngay lệnh của bạn. Nó chuyển câu lệnh qua Catalyst optimizer — một bộ tối ưu truy vấn mở rộng được, dựa trên cả luật (rule-based) lẫn chi phí (cost-based). Catalyst biến ý định của bạn thành một kế hoạch thực thi hiệu quả, qua nhiều giai đoạn:

Đi qua từng bước:

1. Analysis (phân tích): Spark nhận cây logic chưa phân giải (unresolved) — lúc này tên cột amount mới chỉ là chuỗi ký tự. Catalyst tra Catalog để xác nhận cột tồn tại, gắn kiểu dữ liệu, phân giải các tham chiếu. Nếu bạn gõ sai tên cột, lỗi bật ra ở đây (AnalysisException).

2. Logical Optimization (tối ưu logic): Catalyst áp một loạt luật biến đổi cây logic thành optimized logical plan. Ba tối ưu kinh điển:

   • Predicate pushdown (đẩy điều kiện xuống): đẩy phép lọc (filter/WHERE) xuống sát nguồn dữ liệu nhất có thể — thậm chí xuống tận file Parquet — để đọc ít dữ liệu hơn ngay từ đầu.
   • Column pruning (cắt cột thừa): nếu cuối cùng bạn chỉ dùng 3 trên 50 cột, Spark chỉ đọc 3 cột đó (đặc biệt hiệu quả với định dạng cột như Parquet).
   • Constant folding (gộp hằng số): tính trước các biểu thức hằng, ví dụ WHERE 1 = 1 hoặc col + (2 * 3) được rút gọn ngay lúc lập kế hoạch.

3. Physical Planning (lập kế hoạch vật lý): Catalyst sinh nhiều kế hoạch vật lý khả thi (ví dụ join có thể làm bằng broadcast hash join hay sort-merge join) rồi dùng mô hình chi phí cùng thống kê để chọn cái rẻ nhất.

4. Code Generation: kế hoạch vật lý được biên dịch xuống bytecode Java tối ưu bởi Tungsten (mục sau), và cuối cùng thực thi trên nền RDD — vâng, mọi thứ rốt cuộc đều chạy trên RDD ở tầng dưới cùng.

Bây giờ đã rõ nghịch lý: RDD viết tay thường chậm hơn DataFrame vì với RDD, Spark buộc phải chạy đúng lambda hộp đen của bạn, không được sắp xếp lại. Còn với DataFrame, Catalyst tự do đẩy filter, cắt cột, đổi thứ tự phép toán — những tối ưu mà một kỹ sư trung bình khó nhớ áp dụng thủ công cho từng truy vấn. Nói cách khác: khi bạn viết khai báo, bạn cho Spark quyền viết lại code giùm bạn cho nhanh hơn.

Tungsten — bòn rút từng nhịp CPU và byte bộ nhớ

Nếu Catalyst tối ưu kế hoạch, thì Tungsten tối ưu cách thực thi kế hoạch đó ở tầng phần cứng. Ba trụ cột:

• Quản lý bộ nhớ off-heap: thay vì tạo hàng triệu object Java trên heap (tốn bộ nhớ, gây áp lực garbage collector), Tungsten quản lý bộ nhớ nhị phân thô, thường ngoài heap của JVM, với bố cục gọn và ít GC.
• Whole-stage code generation: thay vì diễn giải từng operator riêng lẻ, Tungsten gộp nhiều operator liền kề của một stage thành một hàm Java duy nhất. Ví dụ một filter rồi project được nấu chung thành một vòng lặp, loại bỏ chi phí gọi hàm và cấu trúc trung gian. Kết quả là code chạy nhanh gần bằng code viết tay tối ưu. Trong kế hoạch vật lý, dấu hiệu này hiện ra dưới nhãn WholeStageCodegen.
• Xử lý dạng cột (columnar) và cache-aware: bố cục dữ liệu và thuật toán được thiết kế thân thiện với cache CPU, tận dụng tốt định dạng cột.

Bạn không cần bật Tungsten — nó luôn hoạt động phía sau khi bạn dùng DataFrame/Dataset. Đây là lý do thứ hai (sau Catalyst) khiến API có cấu trúc vượt trội RDD về tốc độ.

Transformation vs Action — và vì sao Spark "lười"

Đây là mô hình tính toán cốt lõi, đúng cho cả RDD lẫn DataFrame. Có hai loại thao tác:

• Transformation (phép biến đổi): tạo ra một dataset mới từ dataset cũ — map, filter, select, groupBy, join... Transformation lười (lazy): gọi chúng không kích hoạt tính toán, chỉ ghi thêm một bước vào kế hoạch (lineage/logical plan).
• Action (hành động): yêu cầu một kết quả cụ thể — count, collect, show, take, write, save... Action mới thực sự kích hoạt Spark chạy toàn bộ chuỗi transformation đã dồn lại.

Sự lười này chính là điều cho phép Catalyst tối ưu: Spark chờ đến khi thấy toàn bộ kế hoạch mới quyết định cách chạy, thay vì thực thi mù quáng từng dòng.

Transformation còn chia làm hai kiểu theo cách dữ liệu di chuyển:

• Narrow transformation (hẹp): mỗi partition đầu vào chỉ đóng góp cho một partition đầu ra — map, filter. Không cần dữ liệu di chuyển giữa các node, rất rẻ.
• Wide transformation (rộng): dữ liệu phải phân phối lại giữa các partition/node — groupBy, join, distinct, orderBy. Việc di chuyển dữ liệu qua mạng này gọi là shuffle, và nó là thao tác đắt nhất trong Spark.

Shuffle là chủ đề lớn đến mức có riêng bài Shuffle & Partitions. Ở đây chỉ cần nhớ: narrow rẻ, wide gây shuffle và cần dè chừng.

# (minh hoạ) — Lazy: không dòng nào chạy cho đến khi gọi action
df = spark.read.parquet("s3://bank/transactions")   # lazy (transformation)
big = df.filter(F.col("amount") > 1_000_000)          # lazy (narrow)
by_ct = big.groupBy("country").sum("amount")          # lazy (wide -> shuffle)

by_ct.show()   # <-- ACTION: đến đây Spark mới thật sự chạy tất cả

Schema — xương sống của DataFrame

Điều làm nên sức mạnh của DataFrame chính là schema: danh sách cột kèm tên và kiểu. Có hai cách để Spark biết schema:

• inferSchema (suy luận): Spark đọc lướt một phần dữ liệu để đoán kiểu. Tiện khi thử nghiệm, nhưng với CSV/JSON lớn nó tốn thêm một lượt quét dữ liệu và đôi khi đoán sai kiểu (ví dụ mã bưu điện 00123 bị đoán thành số, mất số 0).
• Khai báo tường minh: bạn tự định nghĩa StructType. Trong production nên khai báo tường minh — nhanh hơn (bỏ lượt quét suy luận), ổn định, và tránh sai kiểu bất ngờ.

# (minh hoạ) — Khai báo schema tường minh
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, TimestampType
)

schema = StructType([
    StructField("txn_id",   StringType(),    False),
    StructField("amount",   DoubleType(),    True),
    StructField("country",  StringType(),    True),
    StructField("ts",       TimestampType(), True),
])

df = spark.read.schema(schema).csv("s3://bank/txn.csv", header=True)

Một số kiểu dữ liệu Spark hay dùng: StringType, IntegerType, LongType, DoubleType, DecimalType (chính xác cho tiền tệ), BooleanType, TimestampType, DateType, cùng kiểu phức hợp ArrayType, MapType, StructType (lồng nhau).

Đọc/ghi các định dạng

DataFrame đọc và ghi thống nhất qua spark.read / df.write:

• Parquet: định dạng cột, nén tốt, lưu kèm schema, hỗ trợ predicate pushdown và column pruning — mặc định nên dùng cho dữ liệu phân tích.
• CSV: phổ biến nhưng không có schema, không nén sẵn, chậm với dữ liệu lớn — dùng để trao đổi, không để lưu trữ lâu dài.
• JSON: linh hoạt, biểu diễn được dữ liệu lồng nhau, nhưng nặng và chậm parse.

# (minh hoạ)
df = spark.read.parquet("s3://lake/silver/txn")     # đọc parquet (kèm schema)
df.write.mode("overwrite").parquet("s3://lake/gold/agg")

explain() — đọc kế hoạch thực thi

Giống câu EXPLAIN trong SQL, DataFrame có df.explain() để in kế hoạch thực thi — công cụ số một để hiểu Spark thực sự làm gì và vì sao chậm.

# (minh hoạ)
by_ct.explain(mode="formatted")

Cách đọc kết quả (đọc từ dưới lên, vì đó là thứ tự chạy):

• Ở đáy là bước quét nguồn (FileScan parquet). Để ý hai chi tiết vàng: PushedFilters cho biết filter nào đã được đẩy xuống nguồn (predicate pushdown), và danh sách cột được đọc cho thấy column pruning có hoạt động không.
• Xuất hiện Exchange nghĩa là có shuffle — đây là điểm cần soi khi tối ưu.
• Nhãn *(n) WholeStageCodegen (có dấu *) cho biết Tungsten đã gộp các operator thành một hàm — dấu hiệu tốt.
• Với explain(mode="extended"), bạn thấy cả bốn tầng: parsed → analyzed → optimized logical → physical, đúng như sơ đồ Catalyst ở trên.

Tư duy đọc explain() nối thẳng với tư duy EXPLAIN của SQL truyền thống — chi tiết hơn ở bài Spark SQL và Tối ưu Spark.

So sánh với pandas — khi nào cần Spark?

Nhiều người đến từ pandas và băn khoăn khi nào nên đổi. Nguyên tắc gọn:

• pandas: chạy một máy, trong bộ nhớ, tính ngay (eager). Nhanh, đơn giản, hệ sinh thái phong phú — lý tưởng khi dữ liệu vừa RAM một máy (vài GB trở xuống).
• Spark: chạy phân tán, lười, xử lý dữ liệu lớn hơn bộ nhớ một máy — hàng trăm GB đến hàng chục TB — và chịu lỗi tốt.

Chọn Spark khi: dữ liệu không vừa RAM một máy; cần đọc/ghi song song từ data lake; cần tích hợp pipeline phân tán (join nhiều nguồn lớn). Ở lại với pandas khi dữ liệu nhỏ và cần vòng lặp phân tích nhanh — đừng gọi cả một cụm Spark để xử lý file 20MB. (Lưu ý: có Pandas API on Spark cho phép viết cú pháp giống pandas nhưng chạy phân tán, là cầu nối cho ai quen pandas.)

Use case thực tế

Một pipeline ETL ngân hàng điển hình: mỗi đêm đọc hàng chục triệu giao dịch dạng Parquet từ tầng silver của data lake, lọc giao dịch nghi ngờ (amount > ngưỡng), join với bảng khách hàng để lấy phân khúc, rồi tổng hợp theo quốc gia/sản phẩm ghi xuống tầng gold.

• Dùng DataFrame (không phải RDD): để Catalyst tự đẩy filter amount xuống tận file Parquet và chỉ đọc vài cột cần thiết — thay vì nạp toàn bộ vào bộ nhớ.
• Join bảng khách hàng nhỏ với bảng giao dịch lớn: Catalyst có thể chọn broadcast join, tránh shuffle bảng lớn.
• Khai báo schema tường minh cho input để tránh suy luận sai kiểu tiền tệ (DecimalType) và tăng tốc.
• Chạy explain() trước khi đưa lên production để xác nhận có PushedFilters, kiểm tra số lượng Exchange (shuffle) và có WholeStageCodegen không.
• Chỉ rơi xuống RDD nếu gặp logic quá tùy biến (ví dụ parse định dạng log độc quyền) mà hàm dựng sẵn không diễn đạt được.

Ghi nhớ

• Ba tầng API: RDD (điều khiển thấp, không schema, không tối ưu tự động) < DataFrame/Dataset (khai báo, có schema, được Catalyst tối ưu).
• DataFrame = Dataset[Row]. Dataset typed chỉ có ở Scala/Java; PySpark chủ yếu dùng DataFrame.
• Catalyst biến câu lệnh qua: analysis → logical plan → optimized logical plan (predicate pushdown, column pruning, constant folding) → physical plan → RDD. Đây là lý do DataFrame thường nhanh hơn RDD viết tay.
• Tungsten thực thi nhanh nhờ bộ nhớ off-heap, whole-stage codegen và xử lý columnar — luôn bật ngầm với API có cấu trúc.
• Transformation lười, action kích hoạt. Narrow (map, filter) rẻ; wide (groupBy, join) gây shuffle — đắt.
• Schema là xương sống: trong production, khai báo tường minh thay vì inferSchema. Ưu tiên Parquet.
• explain() là công cụ số một để đọc kế hoạch — soi PushedFilters, Exchange, WholeStageCodegen.
• Dữ liệu vừa RAM một máy → pandas; lớn hơn hoặc cần phân tán → Spark.

Đọc tiếp trong series:

• Spark 1 — Tổng quan
• Spark 3 — Spark SQL
• Spark 4 — Shuffle & Partitions
• Spark 5 — Tối ưu Spark