Làm chủ PySpark withColumn() để biến đổi cột DataFrame

Q: Vì sao tác vụ Spark của tôi chạy chậm khi tôi dùng withColumn() quá nhiều?

Khi bạn xâu chuỗi nhiều lần gọi withColumn() , Spark thêm mỗi lần như một bước riêng trong kế hoạch thực thi logic. Theo thời gian, điều này có thể biến thành kế hoạch phình to, khó tối ưu và chạy chậm hơn. Thay vì chồng mười lần withColumn() , hãy thử xây dựng các cột mới trong một lệnh select() duy nhất hoặc dùng withColumns() để thêm nhiều cột trong một lần.

Q: Tôi có thể dùng withColumn() để xóa một cột không?

Không, withColumn() chỉ thêm hoặc thay thế cột, không xóa. Nếu bạn muốn bỏ một cột, hãy dùng drop() . Bạn cũng có thể dùng select() để chỉ giữ lại các cột cần thiết.

PySpark DataFrame là trung tâm khi bạn xây dựng pipeline có khả năng mở rộng trong Spark. Một điều quan trọng cần nhớ là DataFrame là bất biến. Điều này có nghĩa là khi đã có một DataFrame, bạn không thể thay đổi trực tiếp; mỗi khi bạn chỉnh sửa, bạn luôn nhận được một DataFrame mới. Đây là lúc PySpark withColumn() phát huy tác dụng. Nó giúp bạn thêm, cập nhật hoặc chỉnh sửa cột, nhưng vì DataFrame không thay đổi tại chỗ, bạn luôn phải gán lại kết quả vào một biến mới.

Trong hướng dẫn này, tôi sẽ hướng dẫn bạn cách dùng withColumn() để định hình và tinh chỉnh tập dữ liệu của bạn, dù bạn đang xây dựng đặc trưng, dọn dẹp kiểu dữ liệu hay thêm logic.

withColumn() trong PySpark là gì?

Tóm lại, withColumn() trả về một DataFrame mới với một cột được thêm hoặc thay thế. Vì DataFrame không thay đổi, bạn phải gán giá trị trả về đó, df = df.withColumn(...).

Bên dưới, mỗi lần gọi withColumn() sẽ thêm một phép chiếu vào kế hoạch logic. Điều đó ổn nếu bạn chỉ làm một hai lần, nhưng xâu chuỗi quá nhiều có thể làm phình kế hoạch, khiến Spark chạy chậm đi đáng kể.

Mới bắt đầu với PySpark? Bạn có thể nắm vững các nền tảng để xử lý dữ liệu lớn dễ dàng, bằng cách học cách xử lý, truy vấn và tối ưu các tập dữ liệu khổng lồ cho phân tích mạnh mẽ trong khóa học Introduction to PySpark của chúng tôi.

Các cách dùng cốt lõi của PySpark withColumn()

Hãy cùng xem các cách dùng chính của withColumn():

Thêm một cột hằng

Giả sử bạn muốn thêm thời gian nhập liệu hoặc một cờ. Dùng lit() hoặc typedLit() từ pyspark.sql.functions. Ví dụ:

from pyspark.sql.functions import lit
df = df.withColumn("ingest_date", lit("2025-07-29"))Được hỗ trợ bởi

Tạo cột từ dữ liệu hiện có

Có thể bạn muốn giá trị dẫn xuất, nối chuỗi, hoặc tính tổng. Bạn có thể làm như sau:

from pyspark.sql.functions import col, expr
df = df.withColumn("full_name", col("first_name") + expr(" ' ' + last_name"))Được hỗ trợ bởi

Các phép biến đổi số học hoặc dựa trên biểu thức cũng thuộc nhóm này.

Ghi đè một cột hiện có

Nếu bạn đã có một cột và muốn thay đổi nó, withColumn() chỉ việc thay thế:

df = df.withColumn("age", col("age").cast("integer"))Được hỗ trợ bởi

Không cần xóa rồi thêm lại.

Ép kiểu dữ liệu

Thay đổi kiểu của một cột rất đơn giản:

df = df.withColumn("price", col("price").cast("decimal(10,2)"))Được hỗ trợ bởi

Tôi thấy điều này đặc biệt hữu ích khi đọc JSON hoặc CSV “lỏng” nơi các kiểu đến dưới dạng chuỗi.

Nếu bạn muốn thêm ví dụ về PySpark là gì và cách sử dụng nó, tôi gợi ý bài hướng dẫn Getting Started with PySpark của chúng tôi.

Logic điều kiện và biểu thức when()

Đôi khi bạn cần nhiều hơn phép toán đơn giản. Có thể bạn đang xây dựng một cột trạng thái dựa trên điểm số. Hoặc gắn cờ các bản ghi dựa trên tập hợp quy tắc. Đây là lúc when() từ pyspark.sql.functions xuất hiện. Hãy nghĩ nó như một câu lệnh IF. Bạn có thể kết hợp với otherwise() để bao phủ nhiều nhánh.

Ví dụ như sau:

from pyspark.sql.functions import when
df = df.withColumn(
    "grade",
    when(col("score") >= 90, "A")
    .when(col("score") >= 80, "B")
    .when(col("score") >= 70, "C")
    .otherwise("F")
)Được hỗ trợ bởi

Cách viết gần như tiếng Anh: “Nếu điểm ít nhất 90 thì A. Nếu 80 thì B. Tiếp tục… và nếu không khớp cái nào, gán F.” Cách này rất diễn đạt, và Spark sẽ chuyển logic đó thành biểu thức hiệu quả phía sau. Không vòng lặp lồng nhau, không gọi apply(), chỉ là DAG gọn gàng và kế hoạch thực thi rõ ràng.

Điều này hữu ích khi bạn muốn tránh chuyển sang SQL hoặc làm rối mã với UDF.

Bạn có thể học cách thao tác dữ liệu và tạo bộ đặc trưng machine learning trong Spark bằng SQL trong Python từ hướng dẫn Introduction to Spark SQL in Python của chúng tôi.

Biến đổi cột với hàm dựng sẵn và hàm do người dùng định nghĩa

Bạn sẽ thường cần định dạng hoặc cấu trúc lại cột, chuyển chuỗi sang chữ hoa, tách thành phần, nối giá trị, v.v. PySpark có thư viện phong phú các hàm dựng sẵn hoạt động trực tiếp trong withColumn().

Ví dụ:

from pyspark.sql.functions import upper, concat_ws, split
df = df.withColumn("full_caps", upper(col("name")))
df = df.withColumn("city_state", concat_ws(", ", col("city"), col("state")))
df = df.withColumn("first_word", split(col("description"), " ").getItem(0))Được hỗ trợ bởi

Các hàm dựng sẵn rất tuyệt: nhanh, gốc và được tối ưu. Nhưng đôi khi, bạn có quy tắc đặc thù không khớp cái nào. Khi đó, hàm do người dùng định nghĩa (UDF) sẽ hữu ích.

Dùng UDF

Giả sử bạn muốn tính độ dài chuỗi và dán nhãn:

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
def label_length(x):
    return "short" if len(x) < 5 else "long"
label_udf = udf(label_length, StringType())
df = df.withColumn("name_length_label", label_udf(col("name")))Được hỗ trợ bởi

Đủ đơn giản. Nhưng hãy cẩn trọng, UDF có chi phí. Chúng kéo dữ liệu ra khỏi động cơ tối ưu, chạy trong Python, rồi bọc lại. Điều đó ổn khi cần, nhưng với khối lượng lớn, hãy ưu tiên hàm dựng sẵn hoặc biểu thức SQL nếu có thể.

Tìm hiểu cách tạo, tối ưu và sử dụng PySpark UDF, bao gồm Pandas UDF, để xử lý các biến đổi dữ liệu tùy chỉnh hiệu quả và cải thiện hiệu năng Spark trong hướng dẫn How to Use PySpark UDFs and Pandas UDFs Effectively của chúng tôi.

Cân nhắc hiệu năng và thực hành nâng cao

Đến một lúc nào đó, mọi người dùng PySpark đều gặp tình huống: bạn cứ chồng các lời gọi withColumn() và pipeline chậm rì. Lý do? Mỗi lần gọi thêm một lớp vào kế hoạch logic, Spark phải phân tích, tối ưu và thực thi.

Nếu bạn chỉ thêm một hai cột, thì ổn. Nhưng nếu bạn xâu chuỗi năm, sáu hoặc nhiều hơn, bạn nên nghĩ khác đi.

Dùng `select()` khi thêm nhiều cột

Thay vì gọi withColumn() lặp lại, hãy xây dựng danh sách cột mới bằng select():

df = df.select(
    "*",
    (col("salary") * 0.1).alias("bonus"),
    (col("age") + 5).alias("age_plus_five")
)Được hỗ trợ bởi

Cách này xây dựng kế hoạch logic trong một lần.

Tìm hiểu thêm về select() và các phương thức khác trong tài liệu PySpark Cheat Sheet: Spark in Python.

Còn `withColumns()` thì sao?

Được giới thiệu trong Spark 3.3, withColumns() cho phép bạn thêm nhiều cột trong một lượt. Không cần gọi lặp lại. Đây là phương thức kiểu từ điển:

df = df.withColumns({
    "bonus": col("salary") * 0.1,
    "age_plus_five": col("age") + 5
})Được hỗ trợ bởi

Không phải ai cũng dùng Spark 3.3+ rồi, nhưng nếu có, hãy dùng. Gọn hơn, nhanh hơn và tránh vấn nạn “chết vì xâu chuỗi”.

Hướng dẫn ví dụ hoàn chỉnh về PySpark withColumn()

Giả sử bạn đang xử lý dữ liệu hoạt động người dùng cho một nền tảng đăng ký. DataFrame thô trông như sau:

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("withColumn-demo").getOrCreate()
data = [
    ("Alice", "NY", 24, 129.99),
    ("Bob", "CA", 31, 199.95),
    ("Charlie", "TX", 45, 0.0),
    ("Diana", "WA", 17, 19.99)
]
columns = ["name", "state", "age", "purchase_amount"]
df = spark.createDataFrame(data, columns)Được hỗ trợ bởi

Bạn có tên, bang, tuổi và số tiền đã chi. Khá cơ bản. Nhưng trong thực tế, thế là chưa đủ. Vậy tiếp theo ta sẽ làm:

Thêm cột hằng cho ngày nhập liệu.
Tạo cột mới cho biết người dùng là người lớn hay chưa.
Định dạng purchase_amount với hai chữ số thập phân.
Phân loại người dùng theo mức chi tiêu.
Áp dụng hàm tùy chỉnh để dán nhãn người dùng.
Dùng withColumns() để gói thêm các giá trị phụ một cách gọn gàng.

Đang chuẩn bị cho buổi phỏng vấn tiếp theo? Bài viết Top 36 PySpark Interview Questions and Answers for 2025 cung cấp hướng dẫn toàn diện về câu hỏi và câu trả lời phỏng vấn PySpark, bao gồm 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.

Bước 1: Thêm ngày nhập liệu cố định

Theo dõi thời điểm dữ liệu được đưa vào hệ thống là một thói quen tốt.

from pyspark.sql.functions import lit
df = df.withColumn("ingest_date", lit("2025-07-29"))Được hỗ trợ bởi

Bước 2: Gắn cờ người lớn vs. vị thành niên

Bạn có thể chỉ dùng col("age") >= 18, nhưng bọc bằng when() cho bạn toàn quyền nếu logic phức tạp hơn sau này.

from pyspark.sql.functions import when, col
df = df.withColumn(
    "is_adult",
    when(col("age") >= 18, True).otherwise(False)
)Được hỗ trợ bởi

Bước 3: Định dạng purchase_amount

Ép kiểu là một trong những việc dọn dẹp thường xuyên nhất bạn sẽ làm, đặc biệt khi đọc từ CSV hoặc JSON.

df = df.withColumn("purchase_amount", col("purchase_amount").cast("decimal(10,2)"))Được hỗ trợ bởi

Bước 4: Phân loại mức chi tiêu

Giả sử bạn muốn ba nhóm: “none”, “low” và “high”.

df = df.withColumn(
    "spend_category",
    when(col("purchase_amount") == 0, "none")
    .when(col("purchase_amount") < 100, "low")
    .otherwise("high")
)Được hỗ trợ bởi

Điều này giúp bạn phân khúc người dùng mà không cần chạy truy vấn riêng.

Bước 5: Dán nhãn người dùng bằng UDF

Giờ là một quy tắc “bịa” cho ví dụ. Giả sử bạn dán nhãn một người dựa trên độ dài tên.

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
def user_label(name):
    return "simple" if len(name) <= 4 else "complex"
label_udf = udf(user_label, StringType())
df = df.withColumn("label", label_udf(col("name")))Được hỗ trợ bởi

Bước 6: Thêm nhiều cột phụ trong một lần

Có thể bạn muốn thêm vài cột nữa, tuổi theo tháng và lời chào mừng.

df = df.withColumns({
    "age_in_months": col("age") * 12,
    "welcome_msg": col("name") + lit(", welcome aboard!")
})Được hỗ trợ bởi

Gọn gàng hơn nhiều so với gọi withColumn() hai lần.

Dưới đây là DataFrame cuối cùng khi bạn hiển thị:

df.show(truncate=False)Được hỗ trợ bởi

name	state	age	purchase_amount	ingest_date	is_adult	spend_category	label	age_in_months	welcome_msg
Alice	NY	24	129.99	2025-07-29	True	high	complex	288	Alice, welcome aboard!
Bob	CA	31	199.95	2025-07-29	True	high	simple	372	Bob, welcome aboard!
Charlie	TX	45	0.00	2025-07-29	True	none	complex	540	Charlie, welcome aboard!
Diana	WA	17	19.99	2025-07-29	False	low	complex	204	Diana, welcome aboard!

Chuỗi biến đổi kiểu này thường gặp trong kỹ thuật đặc trưng, lập báo cáo hoặc dọn dẹp nguồn dữ liệu bên thứ ba.

Học các nền tảng làm việc với dữ liệu lớn với PySpark từ khóa học Big Data Fundamentals with PySpark của chúng tôi.

Thực tiễn tốt và cạm bẫy của withColumn()

withColumn() của PySpark trông có vẻ đơn giản, nhưng có thể làm khó ngay cả kỹ sư dày dạn nếu bạn không cẩn thận. Dưới đây là những điều có thể âm thầm phá pipeline của bạn, cùng vài thói quen giúp bạn tránh bất ngờ khó chịu.

Luôn gán lại kết quả

Điều này cơ bản nhưng vẫn dễ mắc: withColumn() không sửa DataFrame gốc. Nó trả về một DataFrame mới. Nếu bạn quên gán lại, thay đổi sẽ mất.

df.withColumn("new_col", lit(1))  # This won't do anything
df = df.withColumn("new_col", lit(1))  # This worksĐược hỗ trợ bởi

Cẩn thận ghi đè ngoài ý muốn

Giả sử DataFrame của bạn có cột tên là status. Bạn chạy:

df = df.withColumn("Status", lit("Active"))Được hỗ trợ bởi

Có vẻ vô hại, đúng không? Nhưng Spark mặc định coi tên cột là không phân biệt hoa thường. Nghĩa là bạn vừa ghi đè cột status gốc. Mà không hề hay biết.

Một cách khắc phục là luôn kiểm tra df.columns trước và sau. Hoặc, nếu pipeline cho phép, bật phân biệt hoa thường bằng:

spark.conf.set("spark.sql.caseSensitive", "true")Được hỗ trợ bởi

Đừng dùng literal Python trong biểu thức

Điều này dễ quên. Khi thêm hằng số, tránh dùng giá trị Python thô. Luôn bọc bằng lit().

df = df.withColumn("region", "US")  # Bad
df = df.withColumn("region", lit("US"))  # GoodĐược hỗ trợ bởi

Vì sao? Bởi withColumn() kỳ vọng một biểu thức Column của Spark, không phải giá trị thô. Nếu bạn sơ suất, Spark có thể ném lỗi khó hiểu, hoặc tệ hơn, âm thầm làm hỏng logic phía sau.

Xử lý ngoại lệ bên ngoài `withColumn()`

Đôi khi mọi người sáng tạo và bọc cả khối withColumn() trong try/except. Tốt hơn là cô lập phần rủi ro (như UDF hoặc đọc dữ liệu) và bắt ngoại lệ ở đó. Giữ lớp biến đổi của bạn sạch và dễ dự đoán.

try:
    def risky_udf(x):
        if not x:
            raise ValueError("Empty input")
        return x.lower()
except Exception as e:
    print("Error in UDF definition:", e)Được hỗ trợ bởi

Hãy để Spark thất bại sớm, đừng che giấu sau các khối try lồng nhau.

Tìm hiểu thêm về ngoại lệ trong Python từ hướng dẫn Exception & Error Handling in Python của chúng tôi.

Ưu tiên hàm dựng sẵn thay vì UDF

Đúng là UDF rất mạnh. Nhưng chúng có đánh đổi: hiệu năng chậm hơn, khó gỡ lỗi, và ít tối ưu hóa. Nếu có hàm dựng sẵn làm được việc, hãy dùng nó.

Cách này:

df = df.withColumn("upper_name", upper(col("name")))Được hỗ trợ bởi

Nhanh hơn nhiều so với cách này:

df = df.withColumn("upper_name", udf(lambda x: x.upper(), StringType())(col("name")))Được hỗ trợ bởi

Khi nào không nên dùng withColumn()

Dù withColumn() rất linh hoạt, vẫn có lúc nó không phải công cụ phù hợp.

Bạn đang tái cấu trúc nhiều cột cùng lúc

Nếu bạn thấy mình gọi withColumn() 10 lần liên tiếp, đã đến lúc đổi hướng. Hãy dùng select() và viết các biến đổi như một phép chiếu mới.

df = df.select(
    col("name"),
    col("age"),
    (col("salary") * 0.15).alias("bonus"),
    (col("score") + 10).alias("adjusted_score")
)Được hỗ trợ bởi

Rõ ràng hơn, hiệu năng tốt hơn, và giúp bộ tối ưu của Spark làm việc vì bạn thay vì chống lại bạn.

Bạn muốn viết logic kiểu SQL

Nếu đội của bạn thiên về SQL, và bạn đã đăng ký DataFrame làm view tạm, thường gọn hơn khi chạy câu lệnh SQL.

df.createOrReplaceTempView("users")
df2 = spark.sql("""
    SELECT name, age,
           CASE WHEN age >= 18 THEN true ELSE false END AS is_adult
    FROM users
""")Được hỗ trợ bởi

Điều này có thể dễ dàng hơn cho các nhà phân tích rành SQL hoặc đội ngũ làm việc song song giữa Spark và cơ sở dữ liệu truyền thống.

Xây dựng kỹ năng SQL của bạn với các khóa học, lộ trình và dự án tương tác do chuyên gia thực tiễn tuyển chọn tại SQL Courses.

Bạn đã dùng Spark 3.3+

Nếu bạn dùng Spark 3.3 trở lên và cần thêm nhiều cột, withColumns() là bạn đồng hành. Không chỉ tiện lợi, nó còn có thể nhanh hơn nhờ tạo một cập nhật kế hoạch logic duy nhất.

Học cách triển khai quản lý dữ liệu phân tán và machine learning trong Spark bằng gói PySpark từ khóa học Foundations of PySpark của chúng tôi.

Kết luận

Hàm withColumn() của PySpark là một trong những công cụ đa năng nhất trong bộ biến đổi dữ liệu của bạn, cho phép thêm, chỉnh sửa và thiết kế đặc trưng trực tiếp trong quy trình làm việc xoay quanh DataFrame. Từ ép kiểu và chèn hằng số đến nhúng logic phức tạp với điều kiện và UDF, withColumn() giúp bạn tái định hình dữ liệu lộn xộn thành các pipeline sẵn sàng cho sản xuất.

Nhưng đi kèm sức mạnh là trách nhiệm. Lạm dụng withColumn() trong các chuỗi dài có thể âm thầm làm giảm hiệu năng bằng cách làm phình kế hoạch logic, khiến tác vụ Spark khó tối ưu và gỡ lỗi hơn. Vì vậy biết khi nào nên dùng select(), withColumns(), hoặc thậm chí SQL có thể tạo khác biệt giữa một tác vụ ì ạch và một tác vụ có khả năng mở rộng.

Khi Spark liên tục phát triển, đặc biệt với các tính năng như withColumns() trong Spark 3.3+, hiểu nội bộ và đánh đổi hiệu năng của từng phương thức là chìa khóa để viết mã sạch hơn, nhanh hơn và dễ bảo trì hơn.

Hãy làm chủ các kỹ thuật biến đổi cột quy mô lớn, tránh cạm bẫy phình to kế hoạch, và học cách các chuyên gia tinh gọn pipeline đặc trưng trong khóa học Feature Engineering with PySpark của chúng tôi.

PySpark withColumn() FAQs

Vì sao tác vụ Spark của tôi chạy chậm khi tôi dùng withColumn() quá nhiều?

Khi bạn xâu chuỗi nhiều lần gọi withColumn(), Spark thêm mỗi lần như một bước riêng trong kế hoạch thực thi logic. Theo thời gian, điều này có thể biến thành kế hoạch phình to, khó tối ưu và chạy chậm hơn. Thay vì chồng mười lần withColumn(), hãy thử xây dựng các cột mới trong một lệnh select() duy nhất hoặc dùng withColumns() để thêm nhiều cột trong một lần.

Tôi có thể dùng withColumn() để xóa một cột không?

Không, withColumn() chỉ thêm hoặc thay thế cột, không xóa. Nếu bạn muốn bỏ một cột, hãy dùng drop(). Bạn cũng có thể dùng select() để chỉ giữ lại các cột cần thiết.

Vì sao tôi gặp lỗi khi cố dùng chuỗi hoặc số trong withColumn()?

Điều đó thường xảy ra khi bạn truyền một giá trị Python thô thay vì bọc bằng lit().withColumn() kỳ vọng một biểu thức Column của Spark. Cách đúng là:

from pyspark.sql.functions import lit
df = df.withColumn("new_col", lit(42))  Được hỗ trợ bởi

Tác giả

Derrick Mwiti

Chủ đề

PySpark Python