Ghép mọi thứ đã học thành một pipeline ETL hoàn chỉnh: Extract → Transform → Load, và lưu kết quả ra Parquet một cách hiệu quả. Đây là dạng job bạn sẽ viết hằng ngày trong công việc data thực tế.
Mọi job xử lý dữ liệu, dù lớn hay nhỏ, đều quy về 3 giai đoạn:
Extract — đọc dữ liệu từ nguồn: file CSV/JSON, database qua JDBC, hoặc stream từ Kafka. Spark biến tất cả thành một DataFrame chung.
Transform — phần "ruột" của job: làm sạch (lọc dòng bẩn, chuẩn hóa kiểu dữ liệu), join nhiều bảng, tính toán cột mới, aggregate (group by).
Load — ghi kết quả ra đích: Parquet trên data lake, bảng Delta, hoặc database đích.
.write. Lúc đó Spark mới tối ưu cả chuỗi và thực thi một lần.CSV tiện để xem bằng mắt, nhưng với dữ liệu lớn thì Parquet thắng tuyệt đối:
| Tiêu chí | CSV | Parquet |
|---|---|---|
| Lưu trữ | Text thuần, lưu theo hàng (row) | Nhị phân, lưu theo cột (columnar) + nén |
| Dung lượng | Lớn | Nhỏ hơn ~5–10 lần nhờ nén theo cột |
| Tốc độ đọc | Chậm — luôn phải đọc cả file | Nhanh — đọc đúng cột cần dùng |
| Schema | Không có (phải inferSchema hoặc tự khai) | Có sẵn, nhúng trong file |
| Predicate pushdown | Không | Có — lọc ngay ở tầng đọc file |
Kết luận: trong production, dữ liệu trung gian và đầu ra nên lưu bằng Parquet (hoặc Delta nếu cần ACID/upsert). CSV chỉ dùng cho dữ liệu nguồn hoặc khi cần xuất cho con người đọc.
WHERE price > 100 Spark có thể bỏ qua nguyên khối không thỏa, không cần đọc.Sau khi đã có DataFrame clean đã làm sạch, ghi ra Parquet rất ngắn:
(clean.write .mode("overwrite") .parquet("/home/jovyan/work/data/out/sales_parquet"))
Tham số mode quyết định cách xử lý khi thư mục đích đã tồn tại:
| Mode | Hành vi khi đích đã có dữ liệu |
|---|---|
overwrite | Xóa dữ liệu cũ rồi ghi mới đè lên |
append | Ghi thêm vào, giữ nguyên dữ liệu cũ |
ignore | Nếu đã có dữ liệu thì bỏ qua, không ghi gì |
error / errorifexists | Báo lỗi nếu đích đã tồn tại (mặc định) |
Đọc lại Parquet không cần khai schema vì schema đã nhúng trong file:
df = spark.read.parquet("/home/jovyan/work/data/out/sales_parquet") df.printSchema()
Bạn có thể chia dữ liệu thành thư mục con theo giá trị của một cột:
(clean.write .mode("overwrite") .partitionBy("category") .parquet("/home/jovyan/work/data/out/sales_by_cat"))
partitionBy("category") tạo ra các thư mục con dạng category=Electronics/, category=Books/... Khi sau này bạn query lọc theo category, Spark chỉ đọc đúng thư mục liên quan và bỏ qua phần còn lại — gọi là partition pruning, giúp query nhanh hơn rất nhiều.
customer_id, order_id), bạn sẽ tạo ra hàng nghìn thư mục, mỗi cái chỉ vài dòng — đây là small files problem, làm chậm cả ghi lẫn đọc. Hãy partition theo cột "thưa" như ngày, tháng, danh mục. Khi cần kiểm soát số file đầu ra, dùng coalesce()/repartition() trước khi ghi (học kỹ ở Bài 6).Đây là một job ETL điển hình: đọc bán hàng + khách hàng, làm sạch, join, làm giàu, rồi ghi ra Parquet partition theo tháng.
from pyspark.sql import SparkSession, functions as F spark = (SparkSession.builder .appName("etl") .master("local[*]") .getOrCreate()) # ===== EXTRACT ===== sales = spark.read.csv("/home/jovyan/work/data/sales.csv", header=True, inferSchema=True) customers = spark.read.option("multiLine", True) \ .json("/home/jovyan/work/data/customers.json") # ===== TRANSFORM ===== fact = (sales .filter(F.col("qty") > 0) # bỏ dòng bẩn .withColumn("amount", F.col("qty") * F.col("unit_price")) .withColumn("order_date", F.to_date("order_date", "yyyy-MM-dd")) .join(F.broadcast(customers), "customer_id", "left") # làm giàu .withColumn("year_month", F.date_format("order_date", "yyyy-MM"))) # ===== LOAD ===== (fact.write .mode("overwrite") .partitionBy("year_month") .parquet("/home/jovyan/work/data/out/fact_sales"))
Extract: đọc sales.csv (suy schema tự động) và customers.json (multiLine=True cho JSON nhiều dòng).
Transform: lọc bỏ dòng có qty ≤ 0, tính cột amount, ép order_date về kiểu date, broadcast join với bảng khách hàng nhỏ để làm giàu, rồi tạo cột year_month làm khóa partition.
Load: ghi ra Parquet, đè dữ liệu cũ, chia thư mục theo year_month để các query sau lọc theo tháng được nhanh.
FACT_SALES trên Oracle để hệ thống khác đọc), bạn ghi qua JDBC thay vì ra file.(fact.write .format("jdbc") .option("url", "jdbc:oracle:thin:@host:1521/XEPDB1") .option("dbtable", "FACT_SALES") .option("user", "...") .option("password", "...") .mode("append") .save())
Cùng nguyên tắc với mọi DB khác — chỉ đổi chuỗi url và driver (MySQL, PostgreSQL...).
ojdbc8.jar) vào classpath khi khởi động Spark — qua --jars, spark.jars, hoặc đặt vào thư mục jars. Thiếu driver sẽ báo lỗi No suitable driver.events.jsonl → đếm số event theo type theo từng ngày → ghi ra Parquet, partition theo ngày.sales.csv gốc với thư mục Parquet bạn xuất ra. Quan sát Parquet nhỏ hơn bao nhiêu lần.category và query lọc đúng 1 category. Vào Spark UI quan sát partition pruning — Spark chỉ đọc 1 thư mục thay vì toàn bộ.