PySpark cho người mới bắt đầu: Xây dựng các kỹ năng ở cấp độ trung cấp

Lập trình PySpark cho người mới bắt đầu: Xây dựng kỹ năng cấp độ trung cấp Bước tiếp theo thiết thực đi sâu vào các phân vùng, xáo trộn, kết nối, bộ nhớ đệm và kế hoạch thực thi. Thomas Reid Ngày 10/7/2026 16 phút đọc Chia sẻ Ảnh của AI Nếu bạn đã đọc hai bài viết đầu tiên của tôi trong loạt bài này, bạn đã nắm được khá nhiều kiến thức. Bạn biết cách tạo SparkSession, tải dữ liệu vào DataFrame, định nghĩa lược đồ, làm sạch các cột lộn xộn, kết nối các tập dữ liệu, ghi tệp Parquet và xây dựng các quy trình làm việc PySpark đơn giản. Tôi sẽ đặt liên kết đến các bài viết đó trong loạt bài này ở cuối bài viết này. Chừng đó đã đủ để thực hiện công việc hữu ích. Nhưng một khi dữ liệu của bạn tăng lên hoặc các phép biến đổi trở nên phức tạp hơn, một loạt câu hỏi mới bắt đầu xuất hiện. Tại sao công việc này đột nhiên trở nên chậm? Tại sao một thao tác groupBy() đơn giản lại mất nhiều thời gian hơn dự kiến? Tại sao PySpark lại ghi quá nhiều tệp đầu ra? Khi nào bạn nên lưu vào bộ nhớ đệm một DataFrame? Và shuffle là gì? Bài viết này nói về những bước tiếp theo đó. Chúng ta vẫn giữ mọi thứ thân thiện. Chúng ta sẽ không đi sâu vào điều chỉnh cụm, nội bộ bộ nhớ, Kubernetes, YARN hoặc các cài đặt cấu hình PySpark khó hiểu. Thay vào đó, chúng ta sẽ tập trung vào các ý tưởng thực tế giúp bạn hiểu PySpark làm gì và cách viết mã PySpark hoạt động dễ đoán hơn. Chủ đề bao trùm là: Đạt được cấp độ PySpark trung cấp chủ yếu là hiểu về sự di chuyển của dữ liệu. Một khi ý tưởng đó được hiểu rõ, PySpark bắt đầu trở nên dễ đoán hơn nhiều. Thường có một lý do chính đáng khiến các công việc PySpark của bạn hoạt động kém hiệu quả; bạn chỉ cần biết tìm ở đâu. Khi PySpark bắt đầu chậm Khi bạn mới học PySpark, việc tập trung vào mã là điều tự nhiên. Bạn viết những thứ như: df.filter(df.city == "London") df.groupBy("customer_id").sum("amount") df.join(customers, on="customer_id") Nhưng khi bạn chuyển sang PySpark nâng cao hơn, bạn cần bắt đầu suy nghĩ về những gì PySpark phải làm đằng sau hậu trường. Một số thao tác khá rẻ. Lọc các hàng là một ví dụ điển hình. Spark có thể xem xét từng hàng, đưa ra quyết định có/không và tiếp tục. df_london = df.filter(df.city == "London") Các thao tác khác tốn kém hơn vì PySpark phải di chuyển dữ liệu. Hãy xem xét điều này. Giống như bộ lọc, nó chỉ là một dòng mã, nhưng PySpark phải đảm bảo tất cả các hàng cho cùng một khách hàng kết thúc cùng nhau. df_totals = df.groupBy("customer_id").sum("amount") Đó là một thao tác phức tạp hơn nhiều và có thể cần dữ liệu di chuyển giữa các phần khác nhau của công việc. Sự di chuyển đó thường là nơi phát sinh chi phí. Vì vậy, thói quen cấp độ trung cấp rất đơn giản: Trước khi tìm kiếm các cài đặt điều chỉnh, hãy đặt một câu hỏi đơn giản hơn: dòng mã này có buộc Spark phải di chuyển dữ liệu không? Điều này không có nghĩa là các phép nối, tổng hợp hoặc sắp xếp dữ liệu là xấu. Ngược lại, chúng rất cần thiết, nhưng chúng là loại thao tác bạn nên xử lý cẩn thận hơn một chút. Cách PySpark chia công việc Một PySpark DataFrame không được coi là một khối dữ liệu liền kề duy nhất. PySpark chia nó thành các phân vùng. Bạn có thể coi các phân vùng là các khối dữ liệu mà PySpark có thể xử lý riêng biệt và song song. Đó là một trong những lý do chính khiến PySpark hoạt động hiệu quả và hữu ích cho việc xử lý dữ liệu quy mô lớn. Bạn có thể kiểm tra số lượng phân vùng mà DataFrame có như sau: df.rdd.getNumPartitions() Kết quả bạn nhận được sẽ là một số nguyên đơn giản, ví dụ: 8 Điều đó có nghĩa là PySpark đã chia dữ liệu thành tám khối, và điều này quan trọng vì việc phân vùng ảnh hưởng đến mức độ song song mà PySpark có thể thực hiện. Nếu có quá ít phân vùng, PySpark có thể không sử dụng được tất cả các lõi CPU có sẵn. Nếu có quá nhiều phân vùng, PySpark có thể tốn quá nhiều thời gian để quản lý nhiều tác vụ và tập dữ liệu nhỏ. Cả hai thái cực đều không lý tưởng. Nếu một trong hai tình huống này xảy ra, PySpark cung cấp hai hàm hữu ích để hỗ trợ. Có thể thay đổi số lượng phân vùng bằng hàm `repartition()`: `df_more = df.repartition(16)` Và có thể giảm số lượng phân vùng bằng hàm `coalesce()`: `df_fewer = df.coalesce(4)` Hàm `repartition()` có thể tăng hoặc giảm số lượng phân vùng, nhưng nó khiến PySpark phân phối lại dữ liệu, điều này thường ảnh hưởng đến hiệu suất. Hàm `coalesce()` được sử dụng để giảm số lượng phân vùng và cần ít di chuyển dữ liệu hơn. Một quy tắc thực tế là: Sử dụng `repartition()` khi cần phân tán dữ liệu đồng đều hơn. Sử dụng `coalesce()` khi chỉ muốn ít tệp đầu ra hơn. Ví dụ, giả sử có một tập dữ liệu tự nhiên tạo ra 40 tệp Parquet tương đối nhỏ khi được ghi ra đĩa. Có thể thực hiện như sau: `df.coalesce(4).write.mode("overwrite").parquet("output/sales")` Điều đó cho PySpark biết để ghi kết quả với bốn phân vùng, điều này thường tạo ra bốn tệp đầu ra. Không cần phải quá chú trọng vào số lượng phân vùng chính xác ở giai đoạn này. Điểm quan trọng là các phân vùng kiểm soát cách PySpark chia công việc của nó. Khi đã hiểu những ý tưởng này, PySpark sẽ dễ hiểu hơn nhiều. Các công việc bị chậm lại không còn cảm thấy như những sự kiện ngẫu nhiên và bắt đầu thấy các nguyên nhân có thể xảy ra: một lần xáo trộn ở đây, một lần kết hợp rộng ở đó, quá nhiều tệp nhỏ ở một nơi khác. Khi dữ liệu phải di chuyển Một số hoạt động của PySpark có thể xảy ra bên trong mỗi phân vùng. Những hoạt động khác yêu cầu dữ liệu từ các phân vùng khác nhau được tập hợp lại. Sự di chuyển đó được gọi là xáo trộn (shuffle). Một lần xáo trộn xảy ra khi PySpark phải phân phối lại dữ liệu trên các phân vùng. Đây là một trong những ý tưởng Spark đầu tiên thực sự thay đổi cách viết PySpark. Các lần xáo trộn không phải là hiếm, và chúng không tự động sai, nhưng chúng thường tốn kém về mặt tính toán và nên tránh nếu có thể. Các hoạt động PySpark phổ biến có thể gây ra xáo trộn bao gồm: `groupBy()` `join()` `distinct()` `orderBy()` `repartition()` Lấy ví dụ này: `df.groupBy("customer_id").sum("amount")` Spark không thể tính tổng cuối cùng cho mỗi khách hàng trừ khi tất cả các hàng cho khách hàng đó được tổng hợp. Điều đó thường có nghĩa là di chuyển dữ liệu giữa các phân vùng. Sắp xếp là một ví dụ phổ biến khác: `df.orderBy`