Transform & Làm sạch dữ liệu

Tổng hợp là thao tác cơ bản nhất: gom các dòng theo một (hoặc nhiều) cột rồi tính toán trên từng nhóm — ví dụ doanh thu theo từng category. Trong PySpark, ta dùng groupBy() rồi agg().

from pyspark.sql import functions as F

(sales.groupBy("category")
      .agg(
          F.sum(F.col("qty") * F.col("unit_price")).alias("revenue"),
          F.count("*").alias("orders"),
          F.avg("unit_price").alias("avg_price")
      )
      .orderBy(F.desc("revenue"))
      .show())

groupBy("category") trả về một đối tượng GroupedData — chưa phải DataFrame. Bạn phải gọi tiếp agg() để biến nó về DataFrame. agg() nhận nhiều hàm tổng hợp cùng lúc, và .alias() dùng để đặt tên cột kết quả cho dễ đọc. Cuối cùng orderBy(F.desc(...)) sắp xếp giảm dần theo doanh thu.

Các hàm tổng hợp hay dùng trong module F:

Khác biệt mấu chốt giữa window và groupBy: groupBy gộp nhiều dòng thành một dòng kết quả, còn window giữ nguyên mọi dòng nhưng gắn thêm một giá trị tính trên "khung" (window) xung quanh dòng đó. Nhờ vậy bạn có thể xếp hạng, tính tổng lũy kế, so sánh dòng hiện tại với dòng trước/sau... mà không mất chi tiết.

from pyspark.sql.window import Window

w = Window.partitionBy("category").orderBy(F.desc("amount"))

(df.withColumn("rank", F.rank().over(w))
   .withColumn(
       "running_total",
       F.sum("amount").over(
           w.rowsBetween(Window.unboundedPreceding, Window.currentRow)
       )
   ))

partitionBy("category") chia dữ liệu thành các "khung" theo category (giống groupBy nhưng không gộp). orderBy(F.desc("amount")) sắp thứ tự trong mỗi khung. F.rank().over(w) gán hạng cho từng dòng. Phần rowsBetween(unboundedPreceding, currentRow) định nghĩa khung "từ dòng đầu đến dòng hiện tại" — chính là công thức tính running total (tổng lũy kế).

Các hàm window thường gặp: rank, dense_rank, row_number (xếp hạng), lag/lead (lấy giá trị dòng trước/sau), và mọi hàm tổng hợp (sum, avg...) khi gọi qua .over(w).

Ba hàm xếp hạng khác nhau ở cách xử lý giá trị bằng nhau (tie):

Join dùng để gắn thông tin từ bảng này sang bảng khác qua một cột khóa chung. Ở đây ta ghép sales với customers qua customer_id để biết mỗi đơn hàng thuộc khách nào, vùng nào.

joined = sales.join(customers, on="customer_id", how="inner")
joined.show()

Tham số on là cột (hoặc danh sách cột) khóa chung; how quyết định kiểu join:

Dữ liệu thật luôn có ô trống (null). Spark cho bạn 3 hướng xử lý chính: bỏ dòng, điền giá trị, hoặc lọc/thay thế theo điều kiện.

# bỏ những dòng có bất kỳ null nào
df.na.drop()

# chỉ bỏ dòng null ở một số cột
df.na.drop(subset=["customer_id", "qty"])

# điền giá trị mặc định cho cột qty
df.na.fill({"qty": 0, "region": "unknown"})

# lọc ra các dòng có cột x bị null
df.filter(F.col("x").isNull())

# coalesce: lấy giá trị không-null đầu tiên
df.withColumn("price", F.coalesce(F.col("unit_price"), F.lit(0.0)))

Nhớ rằng sales.csv cố tình có vài dòng bẩn — ví dụ qty âm (không hợp lệ), đơn trùng, kiểu dữ liệu sai. Đây là code làm sạch điển hình, nối nhiều bước transform liền mạch:

clean = (sales
    .filter(F.col("qty") > 0)
    .dropDuplicates(["order_id"])
    .withColumn("unit_price", F.col("unit_price").cast("double"))
    .withColumn("order_date", F.to_date("order_date", "yyyy-MM-dd")))

clean.printSchema()
clean.show(5)

Diễn giải từng bước: filter(F.col("qty") > 0) loại bỏ dòng lỗi qty âm; dropDuplicates(["order_id"]) bỏ đơn trùng (giữ một dòng cho mỗi order_id); cast("double") ép cột giá về kiểu số thực (CSV đọc vào hay là chuỗi); to_date("order_date", "yyyy-MM-dd") parse chuỗi ngày thành kiểu date thực sự để sau này tính toán theo thời gian.

Quy trình làm sạch dữ liệu chuẩn thường gồm các bước:

• Khảo sát — chạy df.summary(), df.printSchema() để hiểu phân bố và phát hiện giá trị lạ.
• Lọc dòng lỗi — bỏ các dòng vi phạm logic nghiệp vụ (qty âm, giá ≤ 0...).
• Khử trùng lặp — dropDuplicates theo khóa định danh (order_id).
• Chuẩn hóa kiểu — cast số, to_date/to_timestamp cho ngày giờ, trim/lower cho chuỗi.
• Xử lý null — drop hoặc fill tùy cột, rồi kiểm tra lại số dòng còn lại.

Dùng sales và customers đã load, tự viết code cho 3 bài sau:

• Doanh thu theo region. Join sales với customers theo customer_id, rồi groupBy("region") và tính tổng qty * unit_price. Sắp xếp giảm dần để xem vùng nào bán mạnh nhất.
• Top 3 đơn mỗi category. Tạo cột amount = qty * unit_price, dùng window partitionBy("category").orderBy(F.desc("amount")) với row_number(), rồi filter giữ những dòng có hạng ≤ 3.
• Đếm khách VIP theo region. Trên bảng customers, lọc vip == True rồi groupBy("region").count(). So sánh với tổng số khách mỗi vùng để ra tỷ lệ VIP.