PySpark cho người mới bắt đầu: Làm chủ những kiến thức cơ bản

Làm việc với dữ liệu trong Python thường bắt đầu với các công cụ như pandas. Chúng trực quan, mạnh mẽ và hoàn hảo cho các tập dữ liệu nhỏ đến vừa phải. Tuy nhiên, ngay khi dữ liệu của bạn tăng lên vượt quá khả năng chứa của bộ nhớ (RAM), các vấn đề về hiệu suất sẽ bắt đầu xuất hiện. Đây chính là lúc PySpark phát huy tác dụng.

Lưu ý rằng trong bài viết này, tôi sẽ thường xuyên sử dụng thuật ngữ Spark và PySpark thay thế cho nhau. Với mục đích của chúng ta, sự khác biệt này không quá quan trọng, nhưng bạn cần nhớ rằng chúng không giống nhau hoàn toàn. Spark là khung tính toán phân tán tổng quát (viết bằng Scala), còn PySpark là API Python chuyên dụng dành cho Spark.

PySpark là gì?

PySpark là API Python cho Apache Spark, một khung tính toán phân tán được sử dụng để xử lý hiệu quả các khối lượng dữ liệu lớn. Thay vì chạy tất cả các tính toán trên một máy duy nhất, Spark phân phối công việc này trên nhiều máy (một cụm máy - cluster), cho phép bạn xử lý dữ liệu ở quy mô lớn trong khi vẫn viết code quen thuộc như người dùng Python.

Một trong những ưu điểm chính của PySpark là nó che giấu đi phần lớn sự phức tạp của các hệ thống phân tán. Bạn không cần quản lý thủ công các luồng (threads), bộ nhớ hay truyền thông mạng. Spark xử lý các vấn đề này giúp bạn, trong khi bạn chỉ cần tập trung mô tả những gì mình muốn làm với dữ liệu thay vì cách thức thực thi.

Nếu bạn hoàn toàn mới bắt đầu với Spark, có ba ý tưởng cốt lõi chính bạn cần học trước khi sử dụng nó:

1. Cụm máy (Clusters)

Khi nghe rằng Spark chạy trên một "cụm máy", điều này có thể nghe có vẻ đáng sợ. Trên thực tế, bạn không cần kiến thức sâu sắc về hệ thống phân tán để bắt đầu. Một cụm máy đơn giản là một nhóm máy chủ kết nối mạng với nhau có thể hợp tác. Trong một ứng dụng Spark chạy trên cụm máy, một máy đóng vai trò là Driver (điều phối công việc), trong khi các máy khác đóng vai trò là Executor (thực hiện tính toán trên các phần dữ liệu). Khi các nút Executor hoàn thành công việc, chúng báo tín hiệu trở lại nút Driver, và Driver có thể thực hiện bất kỳ thao tác nào cần thiết với tập kết quả cuối cùng.

Bạn chỉ cần nhớ rằng, không nhất thiết phải chạy Spark trên một cụm máy tính vật lý. Khi bạn chạy PySpark cục bộ, Spark sẽ mô phỏng một cụm máy trên laptop hoặc PC của bạn bằng cách sử dụng nhiều lõi xử lý. Một trong những điểm mạnh của PySpark là cùng một đoạn code đó sau này có thể được triển khai lên một cụm máy thực tế, dù trên đám mây hay tại chỗ, với rất ít thay đổi.

Sự tách biệt giữa điều phối và thực thi này cho phép Spark mở rộng quy mô. Khi tập dữ liệu tăng lên, nhiều Executor hơn có thể được thêm vào để xử lý dữ liệu song song, giảm thời gian chạy mà không yêu cầu thay đổi code của bạn.

2. Spark DataFrame

Trái tim của PySpark là DataFrame API, cách thức chính để bạn làm việc với dữ liệu trong Spark. DataFrame đơn giản là một bảng dữ liệu, bao gồm các hàng và cột — rất giống với bảng trong cơ sở dữ liệu hoặc DataFrame trong pandas. Nếu bạn đã từng sử dụng SQL hoặc pandas trước đây, các ý tưởng cơ bản sẽ trông rất quen thuộc.

Với Spark DataFrames, bạn có thể thực hiện các tác vụ dữ liệu phổ biến như lọc hàng, chọn cột, nhóm dữ liệu, nối bảng và tính toán các bản tóm tắt như số lượng hoặc giá trị trung bình. Các thao tác này dễ đọc và dễ viết, cho phép bạn tập trung vào những gì mình muốn làm với dữ liệu thay vì các chi tiết kỹ thuật về cách nó chạy.

Điều khiến Spark trở nên đặc biệt là những gì diễn ra "phía sau hậu trường". Spark tự động xác định cách hiệu quả nhất để chạy các thao tác DataFrame của bạn và sau đó thực thi chúng song song trên nhiều máy tính trong cụm. Bạn không cần tự quản lý việc này — Spark xử lý các việc như chia nhỏ dữ liệu, điều phối công việc và khôi phục sau lỗi nếu có sự cố xảy ra.

Nhờ đó, Spark DataFrames có thể xử lý các tập dữ liệu rất lớn, thậm chí những dữ liệu quá lớn để vừa trong bộ nhớ của một máy đơn lẻ. Đồng thời, chúng cung cấp một giao diện đơn giản và quen thuộc, khiến PySpark trở thành một công cụ mạnh mẽ nhưng dễ tiếp cận để làm việc với big data.

3. Đánh giá lười (Lazy) vs. Khẩn trương (Eager)

Một điểm mạnh khác của PySpark đáng để biết là cách tiếp cận đối với thực thi lười (lazy) so với khẩn trương (eager).

Hầu hết các thư viện dữ liệu Python, như Pandas, sử dụng thực thi khẩn trương. Điều này có nghĩa là khi bạn chạy một thao tác, nó thực thi ngay lập tức, tiếp theo là thao tác tiếp theo, v.v.

PySpark xử lý vấn đề này khác nhau bằng cách sử dụng một kỹ thuật gọi là thực thi lười (lazy execution). Khi bạn viết các chuyển đổi dữ liệu, chẳng hạn như chọn cột hoặc lọc hàng, Spark không thực thi chúng ngay lập tức. Thay vào đó, nó xây dựng một kế hoạch thực thi tối ưu hóa và chỉ chạy tính toán khi một hành động (action) (như hiển thị kết quả hoặc ghi dữ liệu vào đĩa) được kích hoạt. Điều này cho phép Spark tối ưu hóa quy trình làm việc trước khi thực thi, làm cho code của bạn hiệu quả hơn mà không cần thêm nỗ lực từ phía bạn.

Để làm rõ điều này, hãy xem xét kịch bản sau. Giả sử chúng ta có một dataframe với 10 triệu bản ghi mà chúng ta muốn:

a) Thêm một cột trống mới gọi là X b) Lọc dữ liệu theo cách nào đó khiến chúng ta loại bỏ 50% bản ghi c) Thực hiện tổng hợp trên các bản ghi còn lại để cột mới X chứa giá trị MAX của một giá trị khác trong hàng đó d) In ra hàng có giá trị X cao nhất

Trên một hệ thống thực hiện thực thi khẩn trương như Pandas, mọi bước đều được thực hiện chính xác như chúng ta đã phác thảo ở trên. Với 10 triệu bản ghi, nó sẽ tạo ra phiên bản mới của dữ liệu cho mỗi bước, gây lãng phí tài nguyên.

Spark, nhờ vào mô hình thực thi lười, không làm bất kỳ công việc nào khi bạn xác định các bước (a), (b), hay (c). Thay vào đó, nó xây dựng một Kế hoạch Logic (Logical Plan). Khi bạn cuối cùng kích hoạt bước (d) — Hành động (Action) — bộ tối ưu hóa của Spark sẽ nhìn vào toàn bộ kế hoạch và nhận ra nó có thể làm việc thông minh hơn nhiều:

• Predicate Pushdown (Đẩy điều kiện lọc xuống): Spark thấy bộ lọc (loại bỏ 50% bản ghi). Thay vì thêm cột X vào 10 triệu hàng, nó di chuyển việc lọc lên ngay từ đầu.
• Tối ưu hóa: Nó chỉ thêm cột X và tổng hợp trên 5 triệu hàng còn lại.
• Kết quả: Nó tránh được việc xử lý 5 triệu bản ghi, tiết kiệm 50% bộ nhớ và thời gian CPU.

Thiết lập môi trường phát triển

Được rồi, đủ lý thuyết rồi. Hãy xem cách bạn có thể cài đặt PySpark trên hệ thống của mình và chạy một số đoạn mã ví dụ. Đối với một bài viết giới thiệu dành cho người mới, việc tạo ra một cụm máy đa nút thực tế nằm ngoài phạm vi của bài viết này. Nhưng như tôi đã đề cập trước đó, Spark có thể tạo một cụm máy tổng hợp trên PC hoặc laptop của bạn nếu nó có đa lõi.

Điều đầu tiên chúng ta sẽ làm là thiết lập một môi trường phát triển riêng biệt cho công việc này, đảm bảo các dự án của chúng ta được cô lập và không can thiệp vào nhau. Tôi đang sử dụng WSL2 Ubuntu cho Windows và Conda cho phần này, nhưng bạn có thể sử dụng bất kỳ môi trường và phương pháp nào bạn quen thuộc.

Cài đặt PySpark, v.v.

# 1. Tạo môi trường mới với Python 3.11 (rất ổn định cho Spark)
conda create -n spark_env python=3.11 -y
# 2. Kích hoạt nó
conda activate spark_env
# 3. Cài đặt PySpark và PyArrow (cần thiết cho tệp Parquet)
pip install pyspark pyarrow jupyter

Để kiểm tra xem PySpark đã được cài đặt đúng chưa, hãy nhập lệnh pyspark vào cửa sổ terminal. Nếu bạn thấy biểu ngữ chào mừng của Spark, mọi thứ đã sẵn sàng.

Ví dụ 1 — Tạo một cụm máy cục bộ

Thực ra điều này khá dễ dàng. Chỉ cần nhập đoạn sau vào notebook của bạn.

from pyspark.sql import SparkSession

# Khởi tạo Spark Session
spark = SparkSession.builder \
    .master("local[*]") \
    .appName("MyLocalCluster") \
    .config("spark.driver.memory", "2g") \
    .getOrCreate()

# Xác minh cụm máy đang chạy
print(f"Spark đang chạy phiên bản: {spark.version}")
print(f"Master URL: {spark.sparkContext.master}")

Khái niệm SparkSession rất quan trọng. Trong những ngày đầu của Spark, người dùng phải xoay sở nhiều "điểm nhập" (như SparkContext cho các chức năng cốt lõi, SQLContext cho dataframe, và HiveContext cho cơ sở dữ liệu). Điều này gây nhầm lẫn cho người mới bắt đầu.

SparkSession được giới thiệu trong Spark 2.0 như một "cửa hàng một điểm" (one-stop shop) cho mọi thứ. Nó là điểm nhập duy nhất để tương tác với các chức năng của Spark.

Ví dụ 2 — Tạo một DataFrame

Tạo DataFrames và thao tác dữ liệu chúng chứa trong PySpark là những gì bạn sẽ làm phần lớn thời gian. Và việc đó khá đơn giản. Ở đây, chúng ta xác định rằng dataframe của chúng ta sẽ chứa ba bản ghi và ba cột được đặt tên.

# 1. Xác định dữ liệu của bạn dưới dạng danh sách các tuple
data = [
    ("Alice", 34, "New York"),
    ("Bob", 45, "London"),
    ("Catherine", 29, "Paris")
]

# 2. Xác định tên cột của bạn
columns = ["Name", "Age", "City"]

# 3. Tạo DataFrame
df = spark.createDataFrame(data, columns)

# 4. Hiển thị kết quả
df.show()

Có khả năng lớn hơn là bất kỳ dataframe nào bạn sử dụng sẽ được tạo ban đầu bằng cách đọc dữ liệu từ tệp hoặc cơ sở dữ liệu. Tạo một tệp CSV tên là sales_data.csv trên hệ thống của bạn với nội dung sau:

transaction_id,customer_name,net_amount,tax_amount, is_member
101,Alice,250.50,25.05,true
102,Bob,120.00,6.00, false
103,Charlie,450.75,25.07,true
104,David,89.99,5.73,false

Tạo một dataframe từ một tệp như thế này rất đơn giản:

# Tải tệp CSV
df = spark.read.format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load("sales_data.csv")

# Hiển thị dữ liệu
print("Nội dung Dataframe:")
df.show()

# Hiển thị các kiểu dữ liệu (Schema)
print("Schema dữ liệu:")
df.printSchema()

Ví dụ 3 — Xử lý dữ liệu

Tất nhiên, một khi bạn có dữ liệu đầu vào trong một dataframe, điều tiếp theo bạn sẽ muốn làm là xử lý hoặc thao tác nó theo một cách nào đó. Điều đó cũng dễ dàng. Nhắc đến sales_data chúng ta vừa tải, giả sử chúng ta muốn tính toán tổng tiền (net + tax) và tỷ lệ thuế dưới dạng phần trăm của tổng tiền cho mỗi bản ghi và thêm chúng vào dataframe ban đầu của chúng ta.

from pyspark.sql import functions as F

# 1. Thêm 'gross_amount' bằng cách cộng net và tax
# 2. Thêm 'tax_percentage' bằng cách chia tax cho tổng tiền mới
df_extended = df.withColumn("gross_amount", F.col("net_amount") + F.col("tax_amount")) \
    .withColumn("tax_percentage",
    (F.col("tax_amount") / (F.col("net_amount") + F.col("tax_amount"))) * 100)

# 3. Tùy chọn: Làm tròn phần trăm đến 2 chữ số thập phân để dễ đọc
df_extended = df_extended.withColumn("tax_percentage", F.round(F.col("tax_percentage"), 2))

# Hiển thị các cột mới cùng với các cột cũ
df_extended.show()

Tóm tắt

Đó là kết thúc chuyến đi ngắn của chúng ta vào thế giới tính toán phân tán với PySpark. Tôi đã giải thích PySpark là gì và tại sao bạn nên cân nhắc sử dụng nó nếu dữ liệu bạn đang xử lý vượt quá giới hạn bộ nhớ. Tóm lại, khả năng mở rộng quy mô của PySpark lên các cụm máy đa nút lớn, mô hình thực thi lười và cấu trúc dữ liệu dataframe khiến nó trở thành một công cụ xử lý dữ liệu lý tưởng.

PySpark được sử dụng rộng rãi trong kỹ thuật dữ liệu, phân tích và các quy trình học máy. Nó tích hợp tốt với các nền tảng đám mây, hỗ trợ nhiều nguồn dữ liệu khác nhau (như CSV, Parquet và cơ sở dữ liệu) và mở rộng quy mô từ laptop đến các cụm máy sản xuất lớn.

Nếu bạn thoải mái với Python và muốn làm việc với các tập dữ liệu lớn mà không phải từ bỏ cú pháp quen thuộc, PySpark là bước tiếp theo tuyệt vời. Nó bắc cầu giữa phân tích dữ liệu đơn giản và xử lý dữ liệu quy mô lớn, biến nó thành một công cụ có giá trị cho bất kỳ ai bước vào thế giới big data.