Hướng dẫn PySpark Filter: Kỹ thuật, Mẹo tối ưu hiệu năng và Tình huống sử dụng

Lọc dữ liệu là thao tác nền tảng trong PySpark, cần thiết để nhanh chóng tinh chỉnh các tập dữ liệu lớn và thu hẹp thông tin liên quan. Sử dụng bộ lọc hiệu quả trong PySpark có thể nâng cao hiệu suất quy trình làm việc cho các kỹ sư dữ liệu tầm trung, nhà khoa học dữ liệu và nhà phát triển xử lý big data.

Bài viết này giới thiệu các kỹ thuật lọc PySpark từ cơ bản đến nâng cao, nêu chiến lược tối ưu hóa để cải thiện hiệu năng, và trình bày các trường hợp sử dụng thực tiễn áp dụng trong bối cảnh đời thực.

Hướng dẫn này giả định bạn đã có một số kiến thức cơ bản về PySpark, nhưng bạn có thể tìm hiểu PySpark là gì và cách sử dụng trong hướng dẫn Bắt đầu với PySpark của chúng tôi.

PySpark Filter là gì?

Như đã đề cập trong hướng dẫn Học PySpark từ con số 0 năm 2025, PySpark là công cụ được sử dụng rộng rãi trong khoa học dữ liệu nhờ tính dễ dùng và hiệu quả. Nó cung cấp các cách trực quan để lọc tập dữ liệu bằng các hàm dựng sẵn như filter() và where(). Những hàm này giúp người làm dữ liệu tách các hàng từ DataFrame thỏa mãn các điều kiện xác định.

Lọc là bước quan trọng trong tiền xử lý dữ liệu, phân tích và tối ưu pipeline. Nó cho phép loại bỏ sớm các bản ghi không liên quan hoặc sai lệch, tiết kiệm thời gian, giảm chi phí và cải thiện các bước phân tích hay mô hình hóa tiếp theo.

Dù đơn giản, áp dụng bộ lọc trong môi trường phân tán có thể gây ra thách thức như nút thắt cổ chai hiệu năng, điều kiện lọc kém hiệu quả hoặc phức tạp, và xử lý chưa đúng dữ liệu thiếu hoặc không hoàn chỉnh.

Hiểu về Spark Filters rất quan trọng để vượt qua phỏng vấn kỹ sư dữ liệu.

Nắm vững các chủ đề chính và câu hỏi trong phỏng vấn big data, từ khái niệm nền tảng như lưu trữ dữ liệu và tính toán phân tán đến các mảng nâng cao như machine learning và bảo mật, với hướng dẫn 30+ Câu hỏi phỏng vấn Big Data hàng đầu của chúng tôi.

Các nguyên lý lọc trong PySpark

Lọc trong PySpark vận hành trong khuôn khổ mô hình thực thi phân tán của nó, xây dựng trên Apache Spark. Khi bạn áp dụng bộ lọc (ví dụ dùng .filter() hoặc .where()), PySpark không quét dữ liệu ngay. Thay vào đó, nó xây dựng một kế hoạch truy vấn logic mô tả các phép biến đổi bạn muốn thực hiện.

Hai khái niệm cốt lõi ảnh hưởng mạnh đến hiệu quả lọc trong PySpark:

• Đánh giá lười (Lazy Evaluation) và
• Catalyst Optimizer.

Đánh giá lười (Lazy evaluation)

PySpark không thực thi lệnh lọc ngay; thay vào đó, nó xây dựng kế hoạch truy vấn logic và hoãn hành động cho đến khi được yêu cầu thực thi. Mô hình này cho PySpark sự linh hoạt lớn hơn để tối ưu hóa các thao tác trên toàn bộ pipeline dữ liệu.

Bộ tối ưu Catalyst

Cỗ máy tối ưu hóa này trong Spark đánh giá các kế hoạch logic, bao gồm lọc, và tự động xác định cách thực thi nhanh nhất, hiệu quả nhất.

Khi bắt đầu thực thi, dữ liệu sau lọc được xử lý song song trên các nút worker bằng RDD hoặc DataFrame. Mỗi partition áp dụng bộ lọc độc lập, giúp khả năng mở rộng cho tập dữ liệu lớn.

Bạn có thể tìm hiểu thêm về RDD trong tài liệu gian lận PySpark: Spark trong Python của chúng tôi.

Cú pháp lọc PySpark cốt lõi

Lọc dữ liệu dựa trên điều kiện khá đơn giản. Bạn có thể truyền điều kiện dưới dạng biểu thức cột hoặc chuỗi giống SQL.

1. Biểu thức cột

Dùng đối tượng cột của DataFrame và các toán tử để định nghĩa điều kiện:

filtered_data = df.filter(df.age > 25)Được hỗ trợ bởi

2. Chuỗi giống SQL

Viết điều kiện dưới dạng chuỗi theo phong cách SQL:

filtered_data_2 = df.filter("age > 25")Được hỗ trợ bởi

Ngoài ra, bạn có thể dùng hàm where() theo cách tương tự:

filtered_data_3 = df.where(df.age > 25)
filtered_data_4 = df.where("age > 25")Được hỗ trợ bởi

Tất cả những cách trên đều hợp lệ và cho kết quả như nhau. Việc lựa chọn thường phụ thuộc vào sở thích cá nhân hoặc phong cách nào dễ đọc hơn trong ngữ cảnh cụ thể của bạn.

Để tìm hiểu thêm về cú pháp SQL trong PySpark, hãy xem khóa học Giới thiệu Spark SQL trong Python của chúng tôi.

Kỹ thuật và ứng dụng lọc trong PySpark

Hãy cùng khám phá các phương pháp lọc khác nhau và trường hợp sử dụng thực tiễn.

Bộ lọc quan hệ

Bạn có thể lọc bằng các toán tử quan hệ như lớn hơn >, nhỏ hơn <, bằng == và các toán tử khác để tách dữ liệu dựa trên ngưỡng số hoặc phân loại:

greater_than_df = df.filter(df.salary > 50000)  
category_match_df = df.filter(df.country == 'France')Được hỗ trợ bởi

Bộ lọc logic

Để kết hợp nhiều điều kiện vào một bộ lọc, các toán tử logic như AND &, OR | và NOT ~ có thể tạo các biểu thức gọn cho yêu cầu phức tạp:

combined_df = df.filter((df.age > 25) & (df.city == 'New York'))Được hỗ trợ bởi

Các hàm lọc nâng cao

PySpark có các hàm nâng cao cho những trường hợp lọc chuyên biệt. Hãy cùng xem một vài hàm.

Kiểm tra thành viên: Lọc các giá trị thuộc một tập xác định sẵn bằng isin:

selected_countries_df = df.filter(df.country.isin('Japan', 'Brazil', 'India'))Được hỗ trợ bởi

Khớp mẫu: Lọc theo mẫu văn bản bằng like cho khớp đơn giản và rlike cho regex:

name_filter_df = df.filter(df.name.like('A%'))  
regex_filter_df = df.filter(df.name.rlike('^[A-Z][a-z]+Được hỗ trợ bởi

Bộ lọc khoảng: Lọc với phương thức tiện lợi như between:

range_df = df.filter(df.age.between(25, 35))Được hỗ trợ bởi

Kiểm tra null: Xử lý giá trị thiếu một cách rõ ràng bằng cách kiểm tra null:

valid_email_df = df.filter(df.email.isNotNull())Được hỗ trợ bởi

Lọc các kiểu dữ liệu phức tạp

Tổ chức thường dùng dữ liệu lồng nhau hoặc có cấu trúc. Lọc trong PySpark hỗ trợ trường lồng và mảng. Với dữ liệu có cấu trúc (struct), bạn lọc trường lồng dễ dàng:

city_df = df.filter(df.address.city == 'Boston')Được hỗ trợ bởi

Với dữ liệu lưu dạng mảng, bạn có thể dùng các hàm dựng sẵn như array_contains:

from pyspark.sql.functions import array_contains  

skill_df = df.filter(array_contains(df.skills, 'Python'))Được hỗ trợ bởi

Kỹ thuật tối ưu hóa để lọc PySpark hiệu quả

Trong phần này, bạn sẽ khám phá các chiến lược để tăng hiệu suất của thao tác lọc trong PySpark.

Predicate pushdown

Predicate pushdown đưa điều kiện lọc xuống gần tầng lưu trữ dữ liệu nhất có thể. Điều này nghĩa là PySpark đọc ít dữ liệu hơn ngay từ đầu, giúp cải thiện hiệu năng.

Ví dụ:

# Read from a Parquet file
df = spark.read.parquet("s3://your-bucket/sales_data/")

# Apply filter on a non-partition column (e.g., product_category)
filtered_df = df.filter(df.product_category == "Electronics")

# Trigger query plan inspection
filtered_df.explain(True)Được hỗ trợ bởi

Trong ví dụ trên, chúng ta áp dụng bộ lọc trên một cột thường. Bình thường, Spark sẽ tải toàn bộ tập dữ liệu rồi mới lọc trong bộ nhớ, nhưng do Parquet hỗ trợ predicate pushdown, Spark có thể gửi điều kiện lọc trực tiếp cho bộ đọc Parquet.

Thay vì đọc tất cả các hàng rồi mới kiểm tra product_category, Spark chỉ đọc các hàng có product_category = 'Electronics'. Điều này giảm I/O đĩa và bộ nhớ, tăng tốc xử lý.

explain in ra kế hoạch truy vấn vật lý, cho thấy chính xác Spark dự định làm gì khi thực thi.

Nếu predicate pushdown được áp dụng, bạn sẽ thấy như sau:

PushedFilters: [EqualTo(product_category,Electronics)]Được hỗ trợ bởi

Điều này xác nhận Spark đã đẩy bộ lọc xuống cấp độ quét tệp thay vì áp dụng sau khi tải dữ liệu.

Predicate pushdown là một trong những tối ưu hóa mạnh mẽ nhất của Spark. Nó:

• Giảm lượng dữ liệu đọc từ đĩa.
• Tăng tốc độ thực thi truy vấn.
• Giảm sử dụng bộ nhớ và I/O mạng.

Bằng cách viết bộ lọc rõ ràng và lưu dữ liệu ở định dạng như Parquet, bạn cho phép Spark tối ưu hóa thực thi bên dưới.

Tối ưu hiệu năng cũng rất quan trọng khi thực hiện join trong PySpark. Tìm hiểu cách tối ưu join, giảm shuffle, xử lý skew và cải thiện hiệu năng trên pipeline big data và quy trình machine learning trong hướng dẫn PySpark Joins: Tối ưu hiệu năng Join cho Big Data của chúng tôi.

Partition pruning

Khi dữ liệu được phân vùng theo các trường liên quan (ví dụ, ngày), PySpark có thể tránh tải các partition không cần thiết. Partition pruning hiệu quả giúp giảm đáng kể việc quét dữ liệu và thời gian thực thi.

Giả sử chúng ta có tập dữ liệu Parquet được phân vùng theo year và month, lưu tại s3://your-bucket/events_data/.

Mỗi partition có dạng:

events_data/year=2023/month=01/
events_data/year=2023/month=02/

...

events_data/year=2024/month=01/Được hỗ trợ bởi

Ví dụ partition pruning trong PySpark có dạng như sau:

df = spark.read.parquet("s3://your-bucket/events_data/")
filtered_df = df.filter((df.year == 2023) & (df.month == 6))

# .explain(True) to verify pruning
filtered_df.explain(True)Được hỗ trợ bởi

year và month là các cột phân vùng, cho phép bộ tối ưu Catalyst của Spark loại bỏ tất cả các partition không liên quan.

Caching và persistence

Kết quả đã lọc thường được dùng nhiều lần trong một quy trình phân tích. Lưu tạm kết quả đã lọc trong bộ nhớ thông qua caching giúp tránh tính toán lặp lại nhưng cần dùng thận trọng để tiết kiệm tài nguyên.

recent_users_df = df.filter(df.registration_date >= '2022-01-01')  

recent_users_df.cache()Được hỗ trợ bởi

Chiến lược lọc nâng cao và tình huống sử dụng

Hãy khám phá các kịch bản phức tạp nơi những kỹ thuật lọc nâng cao được áp dụng.

Lọc theo thời gian

Với dữ liệu chuỗi thời gian, việc lọc thời gian chính xác thường xuyên cần thiết. Lọc dữ liệu trong một khung thời gian cụ thể rất đơn giản:

time_df = df.filter((df.date >= '2023-08-01') & (df.date < '2023-09-01'))Được hỗ trợ bởi

Đoạn trên chọn tất cả bản ghi trong tháng 08/2023. Điều kiện hoạt động với cả cột DateType và TimestampType.

Để phân tích chuỗi thời gian sâu hơn, như tính rolling metrics hoặc xác định xu hướng theo thời gian, bạn có thể kết hợp bộ lọc thời gian với window functions. Ví dụ, bạn có thể lọc theo khoảng ngày rồi áp dụng trung bình trượt hoặc xếp hạng sự kiện trong mỗi ngày:

from pyspark.sql.window import Window
from pyspark.sql.functions import avg
window_spec = Window.partitionBy("date").orderBy("timestamp").rowsBetween(-2, 0)
df_filtered = df.filter((df.date >= '2023-08-01') & (df.date < '2023-09-01'))
df_with_avg = df_filtered.withColumn("rolling_avg", avg("value").over(window_spec))Được hỗ trợ bởi

Ví dụ này tính trung bình trượt 3 hàng mỗi ngày cho các giá trị được sắp theo timestamp. Kết hợp bộ lọc thời gian với window functions giúp bạn kiểm soát chi tiết phân tích theo thời gian.

Lọc dữ liệu streaming

Lọc luồng dữ liệu thời gian thực có những thách thức riêng. Khác với xử lý theo lô, dữ liệu là vô hạn và đến liên tục. Điều này đòi hỏi áp dụng bộ lọc ngay lập tức, đồng thời xử lý cẩn thận các sự kiện đến muộn để đảm bảo kết quả chính xác.

Structured Streaming của PySpark cung cấp các công cụ trực quan để lọc dữ liệu streaming và xử lý vấn đề thời gian hiệu quả. Bạn có thể áp dụng bộ lọc như với DataFrame tĩnh, đồng thời dùng watermark để kiểm soát thời gian hệ thống chờ dữ liệu đến muộn trước khi chốt kết quả.

Ví dụ như sau:

# Example: Real-Time Filtering with Watermarking

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StringType, TimestampType

# Set up Spark session
spark = SparkSession.builder.appName("RealTimeFiltering").getOrCreate()

# Define schema for streaming data
schema = StructType() \
    .add("user_id", StringType()) \
    .add("event_type", StringType()) \
    .add("event_time", TimestampType())

# Read from a Kafka stream (or any supported source)
streaming_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "user_events") \
    .load() \
    .selectExpr("CAST(value AS STRING) as json") \
    .selectExpr("from_json(json, 'user_id STRING, event_type STRING, event_time TIMESTAMP') as data") \
    .select("data.*")

# Apply real-time filter: only 'click' events
filtered_df = streaming_df.filter(col("event_type") == "click")

# Handle late data with watermarking (e.g., 10 minutes tolerance)
filtered_with_watermark = filtered_df \
    .withWatermark("event_time", "10 minutes")

# Write to console (for demo purposes)
query = filtered_with_watermark.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

query.awaitTermination()Được hỗ trợ bởi

Trong ví dụ:

• Lọc thời gian thực được thực hiện với .filter(col("event_type") == "click").
• Structured Streaming xử lý liên tục dữ liệu Kafka đến.
• Watermarking (withWatermark(...)) đảm bảo các sự kiện đến muộn được xử lý trong một khoảng thời gian (10 phút trong ví dụ này).

Kết hợp lọc thời gian thực và watermarking giúp bạn xây dựng ứng dụng streaming đáng tin cậy ngay cả khi có độ trễ mạng hoặc sự kiện bất thường.

Chuẩn bị dữ liệu cho machine learning

Trong pipeline machine learning, lọc hiệu quả rất quan trọng ở giai đoạn tiền xử lý. Nó không chỉ dùng để loại bỏ nhiễu mà còn để chọn ra các đặc trưng phù hợp nhất cho huấn luyện mô hình. Lọc giúp loại bỏ ngoại lệ, bỏ các bản ghi thiếu hoặc không hợp lệ, và giữ lại chỉ những thuộc tính đóng góp tín hiệu có ý nghĩa.

Cách tiếp cận hai hướng này, vừa làm sạch dữ liệu vừa chọn đặc trưng chính, có thể cải thiện đáng kể cả hiệu năng lẫn độ tin cậy của mô hình, như trong khóa học Làm sạch dữ liệu với PySpark của chúng tôi. Nếu không, mô hình có nguy cơ bị lệch do đầu vào không liên quan hoặc bị méo bởi dữ liệu xấu.

Khóa học Big Data với PySpark của chúng tôi trình bày chi tiết hơn về machine learning trong PySpark.

Thực hành tốt và thách thức thường gặp

Trong phần này, chúng ta sẽ tìm hiểu thực hành tốt và những thách thức thường gặp khi lọc dữ liệu với PySpark.

Danh sách kiểm tối ưu hiệu năng

Dưới đây là một số khuyến nghị thực tiễn để nâng cao hiệu quả lọc:

1. Lọc càng sớm càng tốt trong quá trình xử lý.

2. Giới hạn số lượng cột được chọn để giảm data shuffling.

3. Ưu tiên dùng các hàm lọc dựng sẵn của PySpark thay vì UDF (hàm do người dùng định nghĩa).

Những lỗi thường gặp

Tránh các lỗi sau:

1. Điều kiện lọc phức tạp hoặc lồng nhau không cần thiết, làm giảm hiệu quả của bộ tối ưu

Biểu thức lọc quá rườm rà có thể cản trở Catalyst optimizer của Spark. Ví dụ, xâu chuỗi quá nhiều điều kiện OR hoặc các câu lệnh logic lồng sâu có thể làm phức tạp kế hoạch logic, dẫn tới thực thi kém tối ưu. Trong các tình huống có nhiều phép kiểm tra bằng nhau, hãy cân nhắc dùng isin() hoặc thực hiện join với một DataFrame nhỏ hơn để tận dụng broadcast join, hiệu quả hơn cho tập dữ liệu lớn.

2. Lạm dụng caching có thể gây vấn đề bộ nhớ nếu dùng sai

Dù caching hữu ích khi tái sử dụng DataFrame, dùng tràn lan có thể làm cạn kiệt bộ nhớ cụm. Caching các DataFrame lớn hoặc quá nhiều DataFrame mà không đánh giá tần suất tái sử dụng có thể dẫn đến lỗi hết bộ nhớ. Chỉ nên cache những DataFrame được dùng lại nhiều lần và bỏ cache (unpersist) khi không còn cần.

3. Bỏ qua phân bố dữ liệu lệch (skew) có thể làm chậm tốc độ lọc

Data skew, khi một số khóa có lượng dữ liệu vượt trội, có thể gây nút thắt hiệu năng. Các thao tác như join hoặc tổng hợp trên khóa bị skew có thể dẫn tới thời gian thực thi tác vụ không đều. Các kỹ thuật như salting (thêm tiền tố ngẫu nhiên cho khóa), phân vùng tùy chỉnh, hoặc dùng Adaptive Query Execution của Spark có thể giúp giảm thiểu vấn đề này.

Kết luận

Lọc đúng cách đóng vai trò then chốt trong xử lý dữ liệu hiệu quả với PySpark. Nó giúp tinh gọn quy trình phân tích, tiết kiệm tài nguyên tính toán và mang lại sự rõ ràng cho kết quả phân tích tiếp theo. Tận dụng các phương pháp lọc đơn giản nhưng mạnh mẽ của PySpark, cùng với các chiến lược tối ưu như predicate pushdown, partition pruning và caching chọn lọc, sẽ nâng tầm thực hành kỹ thuật dữ liệu của bạn. Áp dụng nhất quán những kỹ thuật này đảm bảo pipeline dữ liệu vững chắc và dễ bảo trì.

Để khám phá thêm về PySpark, hãy xem các khóa học chuyên sâu về Spark của chúng tôi, bao gồm:

• Giới thiệu về PySpark
• Nền tảng PySpark
• Những điều cơ bản về Big Data với PySpark
• Xây dựng Hệ thống Gợi ý với PySpark