Spark 3 — Spark SQL & DataFrame API thực chiến

Từ RDD đến một API khai báo

Ở bài về RDD và DataFrame, bạn đã thấy rằng RDD là API bậc thấp: bạn viết đúng những phép biến đổi cần chạy, còn Spark thi hành gần như y nguyên. Spark SQL đi theo hướng ngược lại — bạn mô tả kết quả mong muốn (khai báo), rồi bộ tối ưu Catalyst tự quyết định cách thực thi hiệu quả nhất. Đây là lý do trong công việc thực tế, gần như toàn bộ code xử lý dữ liệu có cấu trúc đều dùng Spark SQL / DataFrame API thay vì RDD trần.

Điều quan trọng cần nắm ngay: Spark SQL và DataFrame API là hai mặt của cùng một đồng xu. Bạn có thể viết một câu SQL thuần, hoặc gọi chuỗi phương thức DataFrame — cả hai đều được Catalyst dịch về cùng một logical plan, tối ưu như nhau, rồi sinh ra cùng physical plan. Không có bên nào "nhanh hơn" bên kia về bản chất; chọn cái nào dễ đọc hơn cho ngữ cảnh của bạn.

# (minh hoạ) — hai cách viết, cùng một kết quả
# Cách 1: DataFrame API
df_result = (
    df_orders
    .filter(df_orders.status == "PAID")
    .groupBy("customer_id")
    .sum("amount")
)

# Cách 2: Spark SQL trên temp view
df_orders.createOrReplaceTempView("orders")
df_result = spark.sql("""
    SELECT customer_id, SUM(amount) AS total
    FROM orders
    WHERE status = 'PAID'
    GROUP BY customer_id
""")

Lưu ý sandbox: các ví dụ trong bài chạy trên PySpark / Spark SQL (dialect và engine khác PostgreSQL của môi trường này), nên chúng chỉ để minh hoạ cú pháp và ý tưởng, không thực thi tại chỗ. Cú pháp áp dụng cho Spark 3.5+.

createOrReplaceTempView đăng ký một temporary view — một cái tên tạm để câu SQL tham chiếu tới DataFrame. View này chỉ tồn tại trong phiên SparkSession hiện tại và không ghi gì xuống đĩa.

Biến đổi DataFrame cơ bản

DataFrame là tập hợp phân tán các Row, có schema (tên cột + kiểu). Các phép biến đổi đều lazy — chúng chỉ xây dựng kế hoạch, chỉ khi gặp một action (như show, count, write) Spark mới thực sự chạy.

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

df2 = (
    df_orders
    .select("order_id", "customer_id", "amount", "created_at")  # chọn cột
    .where(F.col("amount") > 0)                                  # lọc (alias của filter)
    .withColumn("amount_vnd", F.col("amount") * 1000)           # thêm/ghi đè cột
    .withColumnRenamed("created_at", "order_time")              # đổi tên
    .drop("customer_id")                                        # bỏ cột
    .distinct()                                                  # loại trùng
    .orderBy(F.col("amount_vnd").desc())                        # sắp xếp
    .limit(100)
)

Vài điểm hay nhầm:

• select vs withColumn: select liệt kê tường minh những cột giữ lại; withColumn thêm một cột (hoặc thay cột cùng tên) và giữ nguyên phần còn lại. Nếu bạn withColumn nhiều lần liên tiếp trên vòng lặp lớn, cân nhắc gộp thành một select để plan gọn hơn.
• col("x") vs df["x"] vs df.x: tương đương nhau. Dùng F.col khi cột chưa gắn với DataFrame cụ thể (ví dụ trong biểu thức chung).
• where = filter: hai tên gọi cho cùng một phép toán, chọn tên bạn thích.
• alias: đổi tên một biểu thức, ví dụ F.sum("amount").alias("total").

Aggregation: groupBy, agg, pivot

Tổng hợp là xương sống của phân tích. groupBy gom các dòng theo một hay nhiều khoá, rồi agg áp các hàm tổng hợp.

# (minh hoạ)
summary = (
    df_orders
    .groupBy("customer_id", "status")
    .agg(
        F.sum("amount").alias("total_amount"),
        F.avg("amount").alias("avg_amount"),
        F.count("*").alias("n_orders"),
        F.countDistinct("product_id").alias("n_products"),
        F.collect_list("order_id").alias("order_ids"),   # gom về mảng
    )
)

agg linh hoạt hơn các shortcut như .sum() vì cho phép nhiều hàm và đặt alias trong một lượt. collect_list / collect_set gom giá trị thành mảng (cẩn thận: nếu một nhóm có hàng triệu phần tử, mảng đó có thể làm tràn bộ nhớ executor).

Pivot xoay giá trị của một cột thành các cột riêng — tiện cho báo cáo dạng ma trận:

# (minh hoạ) — doanh thu theo khách hàng x tháng
pivoted = (
    df_orders
    .groupBy("customer_id")
    .pivot("month", ["01", "02", "03"])   # liệt kê giá trị giúp Spark khỏi quét trước
    .agg(F.sum("amount"))
)

Mẹo hiệu năng: luôn truyền danh sách giá trị vào pivot khi biết trước. Nếu bỏ trống, Spark phải chạy một job phụ để tìm các giá trị phân biệt trước khi pivot.

Join: các loại và cạm bẫy

Join là nơi dễ sinh vấn đề hiệu năng nhất, vì nó thường kéo theo shuffle — di chuyển dữ liệu qua mạng để những dòng cùng khoá gặp nhau trên cùng một executor. Chủ đề shuffle được đào sâu ở bài shuffle & partitions; ở đây ta tập trung vào ngữ nghĩa và cách chọn.

# (minh hoạ)
joined = df_orders.join(
    df_customers,
    on="customer_id",       # cột chung -> tự động khử trùng cột khoá
    how="inner",
)

# điều kiện phức tạp
joined = df_orders.join(
    df_customers,
    on=(df_orders.customer_id == df_customers.id) & (df_orders.region == df_customers.region),
    how="left",
)

Các kiểu join thường dùng:

left_semi và left_anti rất hữu ích và hay bị quên: chúng dùng để lọc theo sự tồn tại mà không làm phình cột hay nhân bản dòng như một join thường có thể gây ra.

Broadcast join cho bảng nhỏ

Khi một bảng đủ nhỏ để nằm gọn trong bộ nhớ executor (mặc định ngưỡng spark.sql.autoBroadcastJoinThreshold khoảng 10 MB), Spark có thể broadcast bảng đó tới mọi executor và join cục bộ — không cần shuffle bảng lớn. Đây là một trong những tối ưu đáng giá nhất.

# (minh hoạ) — ép broadcast bảng dim nhỏ
from pyspark.sql.functions import broadcast

joined = df_facts.join(broadcast(df_dim_small), on="dim_id", how="left")

Spark 3 với Adaptive Query Execution (AQE) còn có thể tự chuyển sang broadcast tại runtime khi thấy một bên nhỏ hơn dự đoán — chi tiết ở bài tối ưu. Nhưng gợi ý broadcast() tường minh vẫn giúp khi thống kê không chính xác.

Data skew — kẻ thù thầm lặng

Nếu một khoá join xuất hiện áp đảo (ví dụ customer_id = "GUEST" chiếm 40% dòng), thì partition chứa khoá đó sẽ phình to, một task chạy mãi trong khi các task khác đã xong — gọi là skew. Triệu chứng: 199/200 task xong nhanh, task cuối "treo". Cách xử lý (salting, AQE skew join) được bàn kỹ ở bài shuffle.

Hàm dựng sẵn và window functions

Module pyspark.sql.functions (quy ước import là F) cung cấp hàng trăm hàm chạy native trong JVM/Catalyst — luôn ưu tiên chúng thay vì tự viết logic.

# (minh hoạ) — cột, ngày, chuỗi, điều kiện
enriched = (
    df_orders
    .withColumn("year", F.year("order_time"))
    .withColumn("month", F.date_format("order_time", "yyyy-MM"))
    .withColumn("days_since", F.datediff(F.current_date(), F.col("order_time")))
    .withColumn("email_domain", F.split(F.col("email"), "@").getItem(1))
    .withColumn("upper_name", F.upper(F.trim(F.col("name"))))
    .withColumn(
        "tier",
        F.when(F.col("total_amount") > 1000, "GOLD")
         .when(F.col("total_amount") > 100, "SILVER")
         .otherwise("BRONZE"),
    )
    .fillna({"email_domain": "unknown"})
)

when().when().otherwise() là cách viết CASE WHEN của DataFrame API — dùng cho phân loại có điều kiện.

Window functions

Window functions tính toán trên một cửa sổ dòng liên quan mà không gộp dòng lại (khác groupBy). Đây là công cụ mạnh cho xếp hạng, so sánh với dòng trước/sau, và tổng cộng dồn.

# (minh hoạ)
from pyspark.sql.window import Window

w = Window.partitionBy("customer_id").orderBy(F.col("order_time"))

ranked = (
    df_orders
    .withColumn("rn", F.row_number().over(w))          # 1,2,3... không lặp
    .withColumn("rnk", F.rank().over(w))               # nhảy số khi trùng
    .withColumn("dense", F.dense_rank().over(w))       # không nhảy số
    .withColumn("prev_amount", F.lag("amount", 1).over(w))   # dòng trước
    .withColumn("next_amount", F.lead("amount", 1).over(w))  # dòng sau
    .withColumn(
        "running_total",
        F.sum("amount").over(
            w.rowsBetween(Window.unboundedPreceding, Window.currentRow)
        ),
    )
)

# Lấy đơn mới nhất của mỗi khách:
latest = ranked.filter(F.col("rn") == 1)

Ba điểm cần nhớ:

• row_number luôn cho số duy nhất; rank để lại "lỗ" khi có đồng hạng; dense_rank không để lỗ.
• lag/lead truy cập dòng lân cận trong cùng partition — nền tảng để tính chênh lệch theo thời gian.
• Tổng cộng dồn (running sum) dùng khung rowsBetween(unboundedPreceding, currentRow).

Window functions cũng gây shuffle (theo partitionBy), nên chọn khoá partition hợp lý để tránh skew.

UDF: khi nào cần và vì sao chậm

UDF (User Defined Function) cho phép bạn nhét logic Python tuỳ ý vào một cột. Nghe tiện, nhưng nên là lựa chọn cuối cùng.

# (minh hoạ) — UDF Python thường: TRÁNH nếu có hàm dựng sẵn tương đương
from pyspark.sql.types import StringType

@F.udf(returnType=StringType())
def mask_phone(p):
    if p is None:
        return None
    return p[:3] + "****" + p[-2:]

df2 = df.withColumn("phone_masked", mask_phone(F.col("phone")))

Vì sao UDF Python chậm:

1. Serialization qua ranh giới JVM ↔ Python: mỗi dòng phải được đóng gói (pickle) từ JVM gửi sang tiến trình Python worker, xử lý, rồi gửi ngược lại. Chi phí này lớn và lặp cho từng dòng.
2. Catalyst không nhìn thấy bên trong UDF: với nó, UDF là một "hộp đen", nên không thể tối ưu (không đẩy predicate qua, không hợp nhất phép toán).

Vì vậy quy tắc vàng: luôn tìm hàm dựng sẵn trước. Ví dụ trên có thể viết bằng F.concat(F.substring(...), F.lit("****"), ...) mà không cần UDF.

pandas UDF (vectorized) — nhanh hơn nhiều

Khi buộc phải viết logic Python (ví dụ dùng thư viện chỉ có ở Python), hãy dùng pandas UDF. Chúng chuyển dữ liệu theo lô (batch) qua Apache Arrow — không tuần tự từng dòng — và xử lý cả một pandas.Series một lần, nên nhanh hơn UDF thường nhiều lần.

# (minh hoạ)
import pandas as pd
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType

@pandas_udf(DoubleType())
def zscore(s: pd.Series) -> pd.Series:
    return (s - s.mean()) / s.std()

df2 = df.withColumn("amount_z", zscore(F.col("amount")))

Thứ tự ưu tiên khi cần biến đổi cột: hàm dựng sẵn → pandas UDF → UDF Python thường (chỉ khi không còn cách nào khác).

Đọc và ghi dữ liệu

Vì sao Parquet là mặc định nên dùng

Parquet là định dạng cột (columnar), nén tốt, mang theo schema, và hỗ trợ hai tối ưu quan trọng:

• Column pruning (column pushdown): nếu bạn chỉ select 3 trong 50 cột, Spark chỉ đọc 3 cột đó từ đĩa.
• Predicate pushdown: điều kiện filter được đẩy xuống tầng đọc file, bỏ qua cả khối dữ liệu không thoả — nhờ thống kê min/max trong metadata của Parquet.

CSV/JSON không có hai tối ưu này (phải đọc toàn bộ, rồi mới lọc), nên chậm và tốn hơn nhiều với dữ liệu lớn.

# (minh hoạ) — đọc
df = spark.read.parquet("s3://lake/orders/")

# ghi có phân vùng theo cột
(
    df.write
      .mode("overwrite")
      .partitionBy("year", "month")     # tạo cây thư mục year=.../month=...
      .parquet("s3://lake/orders_out/")
)

partitionBy và partition discovery

partitionBy("year", "month") khi ghi sẽ tạo cấu trúc thư mục kiểu year=2026/month=07/.... Khi đọc lại, Spark tự khám phá partition từ tên thư mục và biến chúng thành cột. Quan trọng: nếu bạn filter(col("year") == 2026), Spark chỉ đọc thư mục tương ứng — gọi là partition pruning — bỏ qua phần còn lại của lake.

Chọn cột phân vùng có lực phân biệt vừa phải: phân vùng theo ngày/tháng thường tốt; phân vùng theo cột có hàng triệu giá trị (như user_id) sẽ tạo hàng triệu thư mục nhỏ — "small files problem" làm chậm metadata và đọc.

Catalog và Hive metastore

Đến giờ ta làm việc với DataFrame và temp view (chỉ sống trong phiên). Để chia sẻ bảng qua các phiên và cho phép truy vấn bằng SQL theo tên, Spark dùng một catalog — thường lưu trong Hive metastore (một CSDL nhỏ chứa metadata: tên bảng, schema, vị trí file, kiểu bảng).

# (minh hoạ)
# Ghi thành bảng có tên trong catalog
df.write.mode("overwrite").saveAsTable("sales.orders")

# Sau đó, ở phiên khác, truy vấn bằng tên:
spark.sql("SELECT COUNT(*) FROM sales.orders").show()

# Duyệt catalog
spark.catalog.listDatabases()
spark.catalog.listTables("sales")

Managed table vs external table — khác biệt cốt lõi:

• Managed (bảng quản lý): Spark/Hive quản lý cả metadata và dữ liệu. DROP TABLE sẽ xoá luôn file dữ liệu. Vị trí do warehouse quyết định.
• External (bảng ngoài): bạn chỉ định LOCATION tới dữ liệu đã có. DROP TABLE chỉ xoá metadata trong catalog, giữ nguyên file. Dùng khi dữ liệu được nhiều hệ thống chia sẻ hoặc bạn muốn kiểm soát vòng đời file riêng.

# (minh hoạ) — external table qua SQL
spark.sql("""
    CREATE TABLE sales.orders_ext (order_id BIGINT, amount DOUBLE)
    USING PARQUET
    LOCATION 's3://lake/orders_out/'
""")

Với các nền tảng lakehouse hiện đại, format bảng còn tiến hoá thành Delta Lake (có ACID, time travel, schema evolution) — chủ đề của bài Delta / lakehouse.

Luồng ETL điển hình

Ghép các mảnh lại, một pipeline Spark SQL thực tế thường có dạng: đọc nhiều nguồn → làm giàu và join → tổng hợp (kể cả window) → ghi ra bảng phân vùng.

Use case thực tế: ETL join bảng lớn + tổng hợp theo cửa sổ

Tình huống: mỗi ngày tính báo cáo hành vi khách hàng. Có bảng orders khổng lồ (hàng tỉ dòng) và bảng dim customers nhỏ (vài trăm nghìn dòng). Cần: gắn thông tin khách vào đơn, đánh dấu đơn đầu tiên của mỗi khách, tính tổng cộng dồn chi tiêu, rồi tổng hợp theo tháng và ghi phân vùng.

# (minh hoạ)
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from pyspark.sql.functions import broadcast

orders = spark.read.parquet("s3://lake/orders/")
customers = spark.read.parquet("s3://lake/dim_customers/")

# 1) Broadcast join bảng dim nhỏ -> tránh shuffle bảng lớn
enriched = orders.join(broadcast(customers), on="customer_id", how="left")

# 2) Window: đơn đầu tiên + chi tiêu cộng dồn theo khách, theo thời gian
w = Window.partitionBy("customer_id").orderBy("order_time")
enriched = (
    enriched
    .withColumn("order_seq", F.row_number().over(w))
    .withColumn("is_first_order", (F.col("order_seq") == 1).cast("int"))
    .withColumn(
        "cumulative_spend",
        F.sum("amount").over(w.rowsBetween(Window.unboundedPreceding, Window.currentRow)),
    )
    .withColumn("month", F.date_format("order_time", "yyyy-MM"))
    .withColumn("year", F.year("order_time"))
)

# 3) Tổng hợp theo (khách, tháng)
report = (
    enriched
    .groupBy("customer_id", "region", "year", "month")
    .agg(
        F.sum("amount").alias("monthly_spend"),
        F.count("*").alias("n_orders"),
        F.max("cumulative_spend").alias("spend_to_date"),
        F.sum("is_first_order").alias("acquired_this_month"),
    )
)

# 4) Ghi phân vùng để truy vấn nhanh về sau
(
    report.write
    .mode("overwrite")
    .partitionBy("year", "month")
    .parquet("s3://lake/reports/customer_monthly/")
)

Vì sao pipeline này "chạy được" ở quy mô lớn:

• broadcast(customers) loại bỏ shuffle trên bảng orders khổng lồ — chỉ gửi bảng dim nhỏ đi khắp cluster.
• Window trên customer_id vẫn cần shuffle theo khoá này; nếu có khách siêu lớn (skew), cần xử lý theo hướng dẫn ở bài shuffle.
• partitionBy("year","month") giúp các truy vấn báo cáo sau này chỉ đọc đúng tháng cần (partition pruning).
• Toàn bộ dùng hàm dựng sẵn, không UDF, nên Catalyst tối ưu tối đa.

Ghi nhớ

• Spark SQL và DataFrame API tương đương — cùng qua Catalyst; chọn theo độ dễ đọc.
• Nắm chắc bộ biến đổi cơ bản: select, where/filter, withColumn, drop, distinct, orderBy, limit.
• groupBy + agg cho tổng hợp; pivot để xoay cột (nhớ liệt kê giá trị).
• Join: hiểu inner/left/right/outer/semi/anti; broadcast bảng nhỏ; cảnh giác skew.
• Ưu tiên hàm dựng sẵn và window functions; UDF Python chậm vì serialization + là hộp đen với Catalyst → nếu cần Python, dùng pandas UDF (vectorized).
• Dùng Parquet để hưởng column pruning + predicate pushdown; ghi partitionBy để có partition pruning, nhưng tránh cột lực phân biệt quá cao (small files).
• Catalog / Hive metastore: phân biệt bảng managed (drop xoá cả data) và external (drop chỉ xoá metadata); dùng saveAsTable để chia sẻ bảng qua các phiên.

Đọc tiếp: RDD & DataFrame · Shuffle & Partitions · Tối ưu Spark · Delta / Lakehouse