PySpark đọc CSV: Tải và xử lý tệp lớn hiệu quả

Các tệp CSV là trụ cột thiết yếu trong xử lý và phân tích dữ liệu. Gần như mọi chuyên gia dữ liệu, từ kỹ sư dữ liệu và nhà khoa học dữ liệu tầm trung đến nhà phát triển sử dụng PySpark, cuối cùng đều phải làm việc với tệp CSV ở một thời điểm nào đó. Tuy nhiên, khi tập dữ liệu tăng từ megabyte lên gigabyte, thậm chí terabyte và hơn thế nữa, các công cụ đơn lẻ, đơn giản như Pandas hoặc thư viện chuẩn của Python sẽ khó đáp ứng tải. Đây là lúc Apache Spark và PySpark trở nên quan trọng để quản lý các tệp CSV quy mô lớn trong môi trường tính toán phân tán.

Bài viết này sẽ cung cấp cho bạn mọi thứ cần biết khi đọc các tệp CSV lớn với PySpark. 

Hãy đảm bảo rằng bạn đã cài đặt PySpark và quen với những điều cơ bản bằng cách làm theo hướng dẫn Bắt đầu với PySpark của chúng tôi. 

Đọc CSV trong PySpark là gì?

PySpark cho phép người dùng đọc các tệp CSV vào các DataFrame phân tán. DataFrame trong PySpark khá giống với Pandas DataFrame, mang đến một giao diện quen thuộc. Tuy nhiên, ở lớp bên dưới, PySpark DataFrame phân tán việc tính toán và lưu trữ trên nhiều nút, đem lại hiệu năng vượt trội cho tập dữ liệu khổng lồ. 

PySpark mang lại những lợi thế thuyết phục cho các tệp CSV lớn, như:

• Nạp dữ liệu phân tán,
• Xử lý giá trị null mạnh mẽ,
• Linh hoạt trong việc định nghĩa schema, và
• Cách thức đơn giản để quản lý nhiều tệp CSV hoặc tệp CSV nén.

Dẫu vậy, vẫn có những thách thức phổ biến, bao gồm xử lý header hiệu quả, suy luận hoặc chỉ định schema chính xác, và quản lý các bản ghi lỗi hoặc không nhất quán. Hãy tiếp tục đọc để khám phá cách xử lý tất cả các thách thức này. 

Đọc tệp CSV trong PySpark là một trong những chủ đề bạn phải nắm vững để vượt qua phỏng vấn PySpark. 36 Câu hỏi và Trả lời phỏng vấn PySpark hàng đầu năm 2025 của chúng tôi cung cấp hướng dẫn toàn diện, bao quát từ khái niệm nền tảng đến kỹ thuật nâng cao và chiến lược tối ưu hóa.  

Những điều cơ bản khi đọc tệp CSV trong PySpark

Đọc dữ liệu CSV thường là một trong những bước đầu tiên và quan trọng nhất trong quy trình PySpark, tạo nền tảng cho các bước biến đổi tiếp theo, phân tích khám phá và các tác vụ học máy. Thực hiện đúng bước này sẽ đảm bảo xử lý dữ liệu sạch hơn và cải thiện hiệu năng cho các bước tiếp theo.

Khung khái niệm

PySpark đọc tệp CSV trong mô hình phân tán của Spark. Thay vì đọc toàn bộ dữ liệu CSV vào bộ nhớ của một máy đơn, Spark chia nhỏ các tác vụ dữ liệu lớn trên nhiều nút của cụm. Bộ tối ưu Catalyst tích hợp của Spark còn cải thiện hiệu năng bằng cách thực thi hiệu quả các thao tác nền tảng cần thiết trong quá trình nạp CSV.

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

Cách đơn giản để đọc tệp CSV là sử dụng các hàm dựng sẵn của Spark: 

spark.read.csv("file_path", header=True, inferSchema=True)Được hỗ trợ bởi

Hoặc tường minh hơn:

spark.read.format("csv").option("header", "True").load("file_path")Được hỗ trợ bởi

Các tham số chính bao gồm:

• file_path - vị trí của các tệp CSV.
• header - đặt tên cột từ header của CSV nếu True.
• inferSchema - tự động suy luận kiểu dữ liệu của cột.
• delimiter - ký tự phân tách cột; mặc định là dấu phẩy.

Hướng dẫn Học PySpark Từ Con Số 0 năm 2025 của chúng tôi đi sâu hơn vào những điều cơ bản của PySpark và cách học hiệu quả. 

Đọc tệp CSV: Tuỳ chọn và cấu hình

PySpark cung cấp nhiều tuỳ chọn chi tiết cho phép bạn kiểm soát tỉ mỉ quá trình đọc CSV.

Header và suy luận schema

Thiết lập header=True yêu cầu Spark sử dụng dòng đầu tiên của CSV làm tên cột.

inferSchema=True  cho phép Spark tự động đoán kiểu cột bằng cách quét dữ liệu của bạn:

spark.read.csv("customers.csv", header=True, inferSchema=True)Được hỗ trợ bởi

Dù suy luận schema tiện lợi và hiệu quả ban đầu, các tập dữ liệu lớn sẽ bị giảm hiệu năng vì Spark phải quét dữ liệu nhiều lần để xác định kiểu dữ liệu.

Chỉ định schema tuỳ chỉnh 

Định nghĩa schema một cách tường minh giúp tăng hiệu năng đáng kể bằng cách loại bỏ các lần quét dữ liệu lặp lại của Spark. Một schema được xác định sẽ truyền đạt trước tên và kiểu cột.

Cách định nghĩa schema tuỳ chỉnh trong PySpark như sau:

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType
schema = StructType([
    StructField("user_id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("score", DoubleType(), True),
])
df = spark.read.csv("customers.csv", schema=schema, header=True)Được hỗ trợ bởi

Tiếp theo, hãy khám phá cách xử lý dấu phân tách trong PySpark khi đọc tệp CSV. 

Xử lý dấu phân tách và ký tự đặc biệt

Nhiều tệp CSV dùng dấu phân tách khác dấu phẩy, như dấu gạch đứng hoặc tab. PySpark cho phép chỉ định rõ dấu phân tách:

spark.read.csv("customers.csv", header=True, delimiter="|")Được hỗ trợ bởi

Ngoài ra, ký tự escape và quote cũng có thể cấu hình để xử lý các tình huống ký tự đặc biệt:

spark.read.csv("data.csv", header=True, escape='\"', quote='"')Được hỗ trợ bởi

Quản lý giá trị null và thiếu

Dữ liệu CSV trong thực tế thường có sự không nhất quán hoặc bản ghi không đầy đủ. PySpark đơn giản hoá việc xử lý null bằng cách chuyển các ký hiệu tuỳ chỉnh thành giá trị null:

spark.read.csv("customers.csv", header=True, schema=schema, nullValue="NA")Được hỗ trợ bởi

Điều này làm rõ các giá trị null, giảm đáng kể việc làm sạch dữ liệu thủ công về sau.

Hãy khám phá các chiến lược khác để xử lý giá trị null. 

Bạn có thể quyết định lọc bỏ các giá trị null: 

# Filter rows where Age is not null
df_filtered = df.filter(df["Age"].isNotNull())
df_filtered.show()Được hỗ trợ bởi

Điều này lọc DataFrame để chỉ giữ các hàng mà cột Age không phải null. Kết quả sẽ trông như sau:

+---+-----+---+------+
| ID| Name|Age|Salary|
+---+-----+---+------+
|  1| John| 25| 50000|
|  3|  Bob| 30|  NULL|
|  4|Carol| 28| 55000|
+---+-----+---+------+Được hỗ trợ bởi

Chiến lược khác là điền giá trị cho các ô null: 

# Replace null values in Age and Salary with default values
df_filled = df.na.fill({"Age": 0, "Salary": 0})
df_filled.show()Được hỗ trợ bởi

Kết quả sẽ như sau: 

+---+-----+---+------+
| ID| Name|Age|Salary|
+---+-----+---+------+
|  1| John| 25| 50000|
|  2|Alice|  0| 60000|
|  3|  Bob| 30|     0|
|  4|Carol| 28| 55000|
|  5|David|  0| 48000|
+---+-----+---+------+Được hỗ trợ bởi

Đọc nhiều tệp và thư mục

PySpark vượt trội trong quản lý các tập dữ liệu lớn với nhiều tệp. Thay vì nạp và gộp tệp trong một thư mục theo tuần tự thủ công, PySpark hỗ trợ mẫu ký tự đại diện để nạp hàng loạt nhanh và hiệu quả:

spark.read.csv("/data/sales/*.csv", header=True, schema=schema)Được hỗ trợ bởi

Cách này gộp hiệu quả nhiều tệp CSV thành một DataFrame trong một thao tác gọn nhẹ.

Khi dữ liệu đã được nạp vào PySpark, các bước tiếp theo bao gồm xử lý dữ liệu, kỹ thuật đặc trưng và xây dựng mô hình học máy. Khóa học Kỹ thuật đặc trưng với PySpark của chúng tôi trình bày chi tiết các khái niệm này. 

Kỹ thuật tối ưu hoá để đọc CSV hiệu quả

Khi xử lý dữ liệu CSV quy mô lớn, tận dụng các chiến lược tối ưu hóa của PySpark là rất quan trọng.

Chiến lược phân vùng

Phân vùng ảnh hưởng lớn đến hiệu năng bằng cách phân phối dữ liệu đồng đều trên các nút trong cụm. Spark cho phép kiểm soát tường minh kích thước và số lượng phân vùng trong quá trình nạp dữ liệu để giúp tăng tốc các thao tác tiếp theo:

df = spark.read.csv("data.csv", header=True, schema=schema).repartition(20)Được hỗ trợ bởi

​​.repartition(20) chia DataFrame thành 20 phân vùng trên cụm Spark của bạn. Vì Spark xử lý dữ liệu theo từng khối, có nhiều phân vùng hơn có thể:

• Cải thiện tính song song
• Cân bằng khối lượng công việc trên cụm
• Tăng tốc các phép biến đổi và ghi dữ liệu

Nếu bạn chạy trên một cụm có nhiều lõi, điều này giúp tận dụng tối đa tài nguyên. Nhưng nếu lạm dụng (ví dụ, 1000 phân vùng cho một tập dữ liệu nhỏ), bạn có thể làm mọi thứ chậm lại.

Khám phá thêm các hàm Spark như repartition với PySpark Cheat Sheet: Spark in Python của chúng tôi. Tài liệu này đi sâu vào khởi tạo Spark trong Python, nạp dữ liệu, sắp xếp và phân vùng lại. 

Bộ nhớ đệm và lưu giữ

Nếu quy trình của bạn lặp đi lặp lại việc truy cập cùng một tập dữ liệu sau khi nạp, lưu DataFrame trong bộ nhớ hoặc trên đĩa có thể cải thiện hiệu năng đáng kể:

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

Tuy nhiên, hãy nhớ rằng việc lưu đệm yêu cầu tài nguyên hệ thống đủ lớn; luôn cân bằng giữa sử dụng bộ nhớ và mức tăng hiệu năng.

Đánh giá lười và kích hoạt hành động

PySpark dựa trên mô hình đánh giá lười: các thao tác trên dataframe biến đổi kế hoạch thay vì thực thi ngay lập tức. Việc đọc tệp thực sự chỉ diễn ra khi cần, được kích hoạt bởi các lệnh như show(), count() hoặc collect():

# no reading yet
df = spark.read.csv("data.csv", header=True, schema=schema)   
# actual read triggered here
df.show(5)  Được hỗ trợ bởi

Tình huống nâng cao và các lưu ý

Hãy khám phá những kịch bản phức tạp hơn mà bạn có thể gặp khi đọc tệp CSV: 

Đọc tệp CSV nén

Spark quản lý hiệu quả các tệp CSV nén như gz hoặc .bz2  một cách trong suốt, không cần cấu hình bổ sung:

spark.read.csv("logs.csv.gz", header=True, schema=schema)Được hỗ trợ bởi

Xử lý bản ghi lỗi

Các tập dữ liệu CSV có thể chứa dòng hỏng. PySpark cung cấp nhiều tuỳ chọn để giúp bạn quản lý lỗi hoặc bản ghi lỗi một cách linh hoạt:

• mode="PERMISSIVE" (mặc định): đưa vào các hàng lỗi với các cột được điền null.
• mode="DROPMALFORMED: bỏ qua lặng lẽ các bản ghi lỗi. Chế độ này không được hỗ trợ trong các hàm dựng sẵn CSV.
• mode="FAILFAST": ném ngoại lệ khi gặp bất kỳ bản ghi lỗi nào.

spark.read.csv("data.csv", header=True, schema=schema, mode="FAILFAST")Được hỗ trợ bởi

Thiết lập locale và mã hóa

Đôi khi, dữ liệu CSV sử dụng mã hóa không tiêu chuẩn. PySpark dễ dàng xử lý các mã hóa khác nhau thông qua tham số encoding:

spark.read.csv("data_utf8.csv", header=True, encoding="UTF-8")Được hỗ trợ bởi

Các mã hóa được hỗ trợ gồm US-ASCII, ISO-8859-1, UTF-8, UTF-16BE, UTF-16LE, và UTF-16. 

Thực tiễn tốt nhất và lỗi thường gặp

Dưới đây là một số thực tiễn tốt nhất cần ghi nhớ khi đọc tệp CSV với Spark: 

• Chỉ định schema tường minh ở bất cứ nơi nào bạn biết cấu trúc tập dữ liệu.
• Kiểm soát phân vùng để phân phối khối lượng công việc hiệu quả.
• Lưu đệm có chiến lược cho các DataFrame được truy cập thường xuyên.
• Kích hoạt hành động đọc một cách có chủ đích, lưu ý đến cơ chế đánh giá lười.

Nhớ tránh: 

• Đặt header=False cho CSV có hàng header.
• Chỉ dựa vào inferSchema với các tập dữ liệu lớn hoặc được truy cập nhiều lần.
• Bỏ qua các thiết lập quan trọng về dấu phân tách hoặc mã hóa.

Kết luận

Hiểu đúng và tận dụng khả năng nạp CSV mạnh mẽ của PySpark là điều thiết yếu để xử lý dữ liệu lớn hiệu quả. Với việc chỉ định schema rõ ràng, xử lý định dạng null tuỳ chỉnh, phân vùng hiệu quả, và quản lý các kịch bản tệp nén hoặc nhiều tệp, quy trình làm việc của bạn sẽ tinh gọn và hiệu năng.

Hãy nhớ rằng dù PySpark mang lại lợi thế to lớn cho các tác vụ dữ liệu quy mô lớn, các công cụ đơn giản hơn như Pandas vẫn có thể đủ dùng cho tập dữ liệu nhỏ. Hãy sử dụng PySpark khi làm việc với dữ liệu vượt quá khả năng của một máy đơn, và luôn cân nhắc lợi thế của tính toán phân tán.

Để khám phá sâu hơn 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ề Dữ liệu lớn với PySpark
• Xây dựng Hệ thống Gợi ý với PySpark