如何使用PySpark为Spark DataFrame生成每行去重的唯一值列

核心思路 #

全程使用Spark原生内置函数完成处理，避免自定义UDF的性能损耗，核心处理分为三步：

1. 用split函数将逗号+空格分隔的字符串拆分为数组
2. 用array_distinct函数对数组内元素去重
3. 用concat_ws函数将去重后的数组重新拼接为逗号+空格分隔的字符串

完整实现代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import split, array_distinct, concat_ws

# 初始化SparkSession
spark = SparkSession.builder.appName("deduplicate_words").getOrCreate()

# 模拟测试数据
test_data = [
    (1, "Peter, Julia, Peter, Michael"),
    (2, None),
    (3, "Michael, Sara, Michael"),
    (4, "John")
]
df = spark.createDataFrame(test_data, schema=["id", "source_value"])

# 核心处理逻辑
df_result = df.withColumn(
    "result_value",
    concat_ws(
        ", ",
        array_distinct(split("source_value", ", "))
    )
)

# 输出结果验证
df_result.select("id", "result_value").show(truncate=False)

输出结果 #

+---+---------------------+
|id |result_value         |
+---+---------------------+
|1  |Peter, Julia, Michael|
|2  |null                 |
|3  |Michael, Sara        |
|4  |John                 |
+---+---------------------+

方案优势 #

• 性能最优：所有逻辑走Spark原生引擎优化，没有Python UDF的进程通信、序列化开销，适配大数据量处理场景
• 逻辑简洁：天然支持NULL值处理，不需要额外编写判空逻辑，三个内置函数即可完成需求
• 适配灵活：如果实际场景中分隔符存在空格数量不统一的情况，只需将split的第二个参数改为正则",\\s*"即可兼容

内容的提问来源于stack exchange，提问作者user2952921