PySpark là gì và tại sao cần xử lý dữ liệu phân tán?

Thay vì xử lý tuần tự từng dòng dữ liệu như các công cụ truyền thống, PySpark phân chia tập dữ liệu lớn thành các phần nhỏ gọi là các phân vùng (partitions) và phân phối chúng trên một cụm máy tính (cluster) để xử lý song song. Điều này giúp giảm thiểu tối đa thời gian tính toán đối với tập dữ liệu có quy mô từ hàng terabyte đến petabyte, đồng thời cung cấp khả năng chịu lỗi (fault-tolerance) tự động thông qua cấu trúc RDD (Resilient Distributed Dataset).

🛠️ Cài đặt môi trường và khởi tạo SparkSession

Trước khi bắt đầu làm việc với dữ liệu phân tán, bạn cần cài đặt thư viện pyspark. Bạn có thể cài đặt nhanh chóng thông qua trình quản lý gói pip tiêu chuẩn của Python:

pip install pyspark

Sau khi cài đặt thành công, bước khởi đầu bắt buộc trong bất kỳ ứng dụng PySpark nào là khởi tạo một SparkSession. Đây là điểm entry-point duy nhất để tương tác với các tính năng nâng cao như Spark SQL, DataFrame API và quản lý cấu hình cụm.

from pyspark.sql import SparkSession

# Khởi tạo SparkSession với tên ứng dụng định danh
spark = SparkSession.builder \
    .appName("BasicPySparkDistributedData") \
    .master("local[*]") \
    .getOrCreate()

# Kiểm tra phiên bản Spark đang chạy trên môi trường
print(f"Phiên bản Spark hiện tại: {spark.version}")

📊 Làm việc với DataFrame cơ bản trong PySpark

DataFrame trong PySpark tương tự như DataFrame trong thư viện Pandas hoặc các bảng cơ sở dữ liệu quan hệ, nhưng có điểm khác biệt cốt lõi là dữ liệu được phân tán và tính toán song song trên cụm.

Bạn có thể tạo một DataFrame từ danh sách dữ liệu có sẵn (in-memory collection) để thử nghiệm nhanh các cấu trúc dữ liệu:

# Định nghĩa dữ liệu mẫu nhân sự
employee_data = [
    (1, "Nguyen Van A", "Engineering", 1200),
    (2, "Tran Thi B", "Marketing", 950),
    (3, "Le Van C", "Engineering", 1350),
    (4, "Pham Thi D", "HR", 800),
    (5, "Hoang Van E", "Marketing", 1100),
    (6, "Vu Thi F", "Engineering", 1400)
]

# Định nghĩa tên các cột tương ứng cho DataFrame
column_names = ["employee_id", "employee_name", "department", "salary"]

# Tạo DataFrame phân tán trong môi trường Spark
employee_df = spark.createDataFrame(employee_data, schema=column_names)

# Hiển thị cấu trúc schema của dữ liệu để kiểm tra kiểu dữ liệu
employee_df.printSchema()

# Hiển thị toàn bộ dữ liệu mẫu mà không bị cắt ngắn nội dung cột
employee_df.show(truncate=False)

💡 Các thao tác xử lý dữ liệu phổ biến (Transformations & Actions)

Xử lý dữ liệu trong PySpark được chia thành hai nhóm lệnh chính vận hành theo cơ chế lười biếng (lazy evaluation): Transformations (các phép biến đổi tạo ra DataFrame mới như filter, select, groupBy nhưng chưa thực thi ngay) và Actions (hành động kích hoạt toàn bộ chuỗi tính toán và trả kết quả về driver như show, count, collect).

Dưới đây là ví dụ thực hiện các thao tác lọc dữ liệu và tổng hợp lương trung bình theo phòng ban:

from pyspark.sql.functions import avg, col

# Lọc danh sách nhân viên thuộc phòng Engineering có mức lương lớn hơn 1200
filtered_employee_df = employee_df.filter((col("department") == "Engineering") & (col("salary") > 1200))

print("Nhân viên phòng Engineering có lương > 1200:")
filtered_employee_df.show()

# Tính lương trung bình của từng phòng ban sử dụng hàm tổng hợp
average_salary_df = employee_df.groupBy("department") \
    .agg(avg("salary").alias("average_salary"))

print("Thống kê lương trung bình theo từng phòng ban:")
average_salary_df.show()

Khi đã hoàn thành xong các tác vụ tính toán và lưu trữ kết quả, việc giải phóng tài nguyên hệ thống là rất quan trọng để tránh quá tải bộ nhớ trên cụm:

# Đóng kết nối SparkSession và giải phóng tài nguyên cụm
spark.stop()

Nắm vững các kỹ thuật khởi tạo môi trường, xây dựng DataFrame phân tán và phân định rõ ràng giữa lệnh biến đổi và hành động sẽ tạo ra bước đệm vững chắc giúp bạn tự tin chinh phục các bài toán dữ liệu lớn phức tạp hơn trong hệ sinh thái Big Data.

🏋️ Bài tập thực hành 1: Đọc tệp dữ liệu CSV và xử lý dữ liệu thiếu

Mô tả yêu cầu: Tạo một tệp dữ liệu giả lập dạng CSV hoặc sử dụng DataFrame để mô phỏng bài toán đọc file. Thực hiện việc đếm số lượng dòng dữ liệu có chứa giá trị trống (null) và thay thế chúng bằng giá trị mặc định.
Mã nguồn gợi ý:

from pyspark.sql.functions import col, when

# Tạo DataFrame mô phỏng dữ liệu bán hàng có chứa giá trị thiếu (null)
sales_data = [
    (101, "Laptop", 2, 1200.0),
    (102, "Mouse", None, 25.0),
    (103, "Keyboard", 5, None),
    (104, "Monitor", 1, 300.0)
]
sales_columns = ["order_id", "product_name", "quantity", "price"]

sales_df = spark.createDataFrame(sales_data, schema=sales_columns)

# Kiểm tra và đếm số lượng giá trị null trong từng cột
print("Số lượng giá trị null trên mỗi cột:")
for column_name in sales_df.columns:
    null_count = sales_df.filter(col(column_name).isNull()).count()
    print(f"- {column_name}: {null_count} dòng")

# Điền giá trị mặc định: quantity = 1 (nếu null), price = 50.0 (nếu null)
cleaned_sales_df = sales_df.fillna({"quantity": 1, "price": 50.0})

print("Dữ liệu sau khi xử lý giá trị thiếu:")
cleaned_sales_df.show()

🏋️ Bài tập thực hành 2: Thêm cột tính toán theo điều kiện và sắp xếp dữ liệu

Mô tả yêu cầu: Dựa trên DataFrame nhân sự (employee_df) đã tạo ở bài viết, hãy sử dụng hàm withColumn kết hợp với biểu thức điều kiện để tính toán cột tiền thưởng (bonus). Nếu lương lớn hơn 1200, thưởng 200; ngược lại thưởng 100. Sau đó, sắp xếp kết quả theo thứ tự lương giảm dần.
Mã nguồn gợi ý:

from pyspark.sql.functions import col, when

# Thêm cột bonus dựa trên điều kiện mức lương hiện tại
bonus_employee_df = employee_df.withColumn(
    "bonus",
    when(col("salary") > 1200, 200).otherwise(100)
)

# Sắp xếp danh sách nhân viên theo mức lương giảm dần (descending)
sorted_employee_df = bonus_employee_df.orderBy(col("salary").desc())

print("Danh sách nhân viên sau khi tính tiền thưởng và sắp xếp theo lương giảm dần:")
sorted_employee_df.show(truncate=False)