PySpark cơ bản & Kiến trúc

Spark là một engine xử lý dữ liệu phân tán: thay vì một máy làm hết, công việc được chia nhỏ và chạy song song trên nhiều tiến trình. Có 4 nhân vật chính bạn cần phân biệt:

• Driver — tiến trình chạy chương trình PySpark của bạn. Nó giữ SparkSession, phân tích code thành kế hoạch thực thi, chia thành các task và gửi đi. Driver là "bộ não" điều phối, nhưng không tự xử lý dữ liệu.
• Executor — tiến trình thực sự chạy các task và giữ dữ liệu trong RAM (cache). Mỗi executor có nhiều core, mỗi core chạy một task tại một thời điểm. Đây là nơi "cơ bắp" xử lý diễn ra.
• Worker — máy (node) trong cụm, là nơi các executor được khởi chạy. Một worker có thể chứa một hoặc nhiều executor.
• Cluster Manager — bộ phận cấp phát tài nguyên (CPU, RAM) và quyết định khởi chạy executor ở đâu. Có thể là Spark Standalone, YARN, Kubernetes... Ở chế độ local thì vai trò này được mô phỏng ngay trong tiến trình của bạn.

Luồng đi của một job nhìn đơn giản như sau:

Điểm mấu chốt: dữ liệu được chia thành nhiều partition (phân vùng). Mỗi partition là một mẩu dữ liệu được một task xử lý độc lập. Nhờ vậy, nếu có 6 partition và 3 core rảnh, Spark xử lý 3 partition cùng lúc — đó chính là tính song song giúp Spark nhanh trên dữ liệu lớn.

SparkSession là điểm vào duy nhất của mọi chương trình Spark hiện đại. Qua nó bạn đọc dữ liệu, tạo DataFrame, chạy SQL... Tạo session như sau:

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("bai1")
    .master("local[*]")
    .getOrCreate()
)

Giải nghĩa từng phần:

• builder — đối tượng dạng "thợ xây", cho phép bạn cấu hình session bằng cách nối chuỗi các phương thức rồi kết thúc bằng getOrCreate().
• appName("bai1") — đặt tên cho ứng dụng. Tên này hiện trên Spark UI, giúp bạn nhận ra job của mình khi có nhiều ứng dụng.
• master("local[*]") — chỉ định nơi chạy. local[*] = chạy local dùng hết core. Nếu submit lên cụm thật, chỗ này sẽ là URL của cluster manager (vd spark://host:7077).
• getOrCreate() — nếu đã có session thì dùng lại, chưa có thì tạo mới. Nhờ vậy chạy lại cell nhiều lần cũng không tạo trùng session.

DataFrame là một bảng dữ liệu phân tán có schema (tên cột + kiểu dữ liệu) — giống bảng trong SQL hay DataFrame của pandas, nhưng khác ở hai điểm quan trọng: nó được chia thành nhiều partition nằm rải trên các executor, và nó lazy (lười) — chỉ ghi nhận kế hoạch chứ chưa thực thi cho đến khi gặp một action. Ta đọc CSV:

sales = spark.read.csv(
    "/home/jovyan/work/data/sales.csv",
    header=True,
    inferSchema=True,
)

sales.printSchema()   # in cấu trúc cột + kiểu
sales.show(5)         # in 5 dòng đầu

• header=True — coi dòng đầu của CSV là tên cột (thay vì dữ liệu).
• inferSchema=True — bảo Spark tự đoán kiểu mỗi cột (số nguyên, số thực, chuỗi...). Rất tiện, nhưng chậm vì Spark phải quét file hai lần: một lần đoán kiểu, một lần đọc thật. Với dữ liệu lớn ở production, nên khai báo schema bằng tay.

Đọc JSON — lưu ý có hai dạng khác nhau:

# JSON dạng "đẹp" (một array trải nhiều dòng) → cần multiLine
customers = (
    spark.read
    .option("multiLine", True)
    .json("/home/jovyan/work/data/customers.json")
)

# JSON-lines: mỗi dòng là một object JSON độc lập → KHÔNG cần multiLine
events = spark.read.json("/home/jovyan/work/data/events.jsonl")

Sự khác biệt: file JSON array là một cấu trúc lớn trải dài nhiều dòng, nên cần multiLine=True để Spark hiểu cả file là một khối. Còn JSON-lines (.jsonl) là mỗi dòng một object JSON hoàn chỉnh — định dạng này lý tưởng cho dữ liệu lớn vì Spark đọc và chia partition theo dòng rất hiệu quả, không cần multiLine.

from pyspark.sql.types import (
    StructType, StructField,
    IntegerType, StringType, DoubleType,
)

schema = StructType([
    StructField("order_id",   IntegerType(), True),
    StructField("category",   StringType(),  True),
    StructField("qty",        IntegerType(), True),
    StructField("unit_price", DoubleType(),  True),
])

sales = spark.read.csv(
    "/home/jovyan/work/data/sales.csv",
    header=True, schema=schema,
)

Để xử lý cột, ta dùng module hàm dựng sẵn của Spark, theo quy ước đặt bí danh là F:

from pyspark.sql import functions as F

df = (
    sales
    .withColumn("amount", F.col("qty") * F.col("unit_price"))
    .filter(F.col("qty") > 0)
    .select("order_id", "category", "amount")
)

df.show(5)

• F.col("qty") — tham chiếu tới cột qty dưới dạng biểu thức. Bạn có thể tính toán trên nó: nhân, cộng, so sánh... Ở đây F.col("qty") * F.col("unit_price") tạo biểu thức "số lượng × đơn giá".
• withColumn("amount", ...) — thêm một cột mới tên amount (hoặc ghi đè nếu đã tồn tại) bằng giá trị của biểu thức.
• filter(F.col("qty") > 0) — giữ lại các dòng thỏa điều kiện (ở đây loại bỏ dòng có số lượng ≤ 0, tức dòng bẩn).
• select("order_id", "category", "amount") — chọn ra đúng những cột cần, bỏ phần còn lại cho gọn.

Quan trọng: cả ba withColumn, filter, select đều là transformation — chúng chỉ mô tả bạn muốn biến đổi dữ liệu ra sao và trả về một DataFrame mới. Chưa có dòng dữ liệu nào được xử lý cho đến khi gọi show() ở cuối (đó mới là action).

Đây là khái niệm quan trọng nhất của bài. Spark chia mọi thao tác làm hai loại:

• Transformation (lười) — chỉ ghi lại kế hoạch, không chạy gì cả. Mỗi transformation tạo ra một DataFrame mới mô tả "sẽ làm gì", nối thành một chuỗi gọi là lineage (lý lịch). Gọi 100 transformation liên tiếp vẫn trả về tức thì vì chưa hề đụng tới dữ liệu.
• Action (chạy thật) — khi gặp một action, Spark mới thực sự khởi chạy job: gửi task xuống executor, đọc và xử lý dữ liệu, rồi trả kết quả về. Không có action thì không có gì xảy ra.

Muốn xem Spark định chạy thế nào mà không thật sự chạy job, dùng explain() để in ra physical plan:

df.explain()        # in physical plan, KHÔNG chạy job
df.explain(True)    # in cả logical + physical plan chi tiết hơn

Bạn sẽ thấy một cây thực thi mô tả Spark sẽ quét file, lọc, tính cột... theo thứ tự nào — nhưng không có dòng dữ liệu nào bị xử lý.

Mở notebook và thử ba bài sau (gợi ý đã có trong nội dung phía trên):

• Đếm đơn thanh toán bằng thẻ có giá trị lớn: từ df (đã có cột amount), đếm số đơn vừa thỏa payment == "card" vừa có amount > 1000. Gợi ý: dùng filter với điều kiện kết hợp (F.col("payment") == "card") & (F.col("amount") > 1000) rồi .count().
• Top 10 đơn giá trị cao nhất: sắp xếp df theo amount giảm dần rồi lấy 10 dòng đầu. Gợi ý: df.orderBy(F.desc("amount")).show(10).
• Đếm dòng bẩn: trong sales gốc (chưa lọc), đếm số dòng có qty < 0 để biết dữ liệu bẩn nhiều cỡ nào. Gợi ý: sales.filter(F.col("qty") < 0).count().