PySpark从值列表添加列问题：自定义UDF返回重复值求助

首先咱们得搞清楚你原来的方法为啥行不通：你在UDF里用了rating.pop(0)，但Spark是分布式计算框架，UDF会在多个executor上独立执行，每个executor都会单独加载这个评分列表并执行pop操作，而且全局列表的修改在分布式环境里是不共享的——这就导致所有行都拿到了列表的第一个值，完全达不到按顺序匹配的效果。

下面给你两种靠谱的解决方案：

方法一：添加行号后与评分列表关联 #

这种方法直观易懂，适配绝大多数场景：

1. 给原DataFrame添加连续行号（保证和原数据行顺序一致）
2. 把评分列表转换成带行号的小DataFrame
3. 通过行号将两个DataFrame关联，再去掉临时行号列

代码示例：

from pyspark.sql import SparkSession
from pyspark.sql.functions import monotonically_increasing_id
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

# 初始化SparkSession（如果未初始化的话）
spark = SparkSession.builder.appName("AddRatingColumn").getOrCreate()

# 原DataFrame
a = spark.createDataFrame([("Dog", "Cat"), ("Cat", "Dog"), ("Mouse", "Cat")],["Animal", "Enemy"])
rating = [5,4,1]

# 给原DataFrame添加连续行号
window = Window.orderBy(monotonically_increasing_id())
df_with_row_num = a.withColumn("row_num", row_number().over(window))

# 将评分列表转为带行号的DataFrame（行号从1开始，和上面的row_number对应）
rating_df = spark.createDataFrame([(i+1, score) for i, score in enumerate(rating)], ["row_num", "Rating"])

# 关联并清理临时列
new_df = df_with_row_num.join(rating_df, on="row_num", how="inner").drop("row_num")

# 查看结果
new_df.show()

执行后就能得到你期望的结果：

+------+-----+------+
|Animal|Enemy|Rating|
+------+-----+------+
|   Dog|  Cat|     5|
|   Cat|  Dog|     4|
| Mouse|  Cat|     1|
+------+-----+------+

方法二：利用RDD的zipWithIndex（适合熟悉RDD的场景） #

如果你对RDD比较熟悉，可以用更简洁的方式实现：先把DataFrame转成RDD，用zipWithIndex添加索引，再将索引和评分列表匹配后转回DataFrame：

# 原DataFrame转RDD并添加索引（索引从0开始）
rdd_with_index = a.rdd.zipWithIndex()

# 映射索引到评分列表，再转回DataFrame
new_rdd = rdd_with_index.map(lambda x: (x[0][0], x[0][1], rating[x[1]]))
new_df = spark.createDataFrame(new_rdd, ["Animal", "Enemy", "Rating"])

new_df.show()

这种方法更简洁，但要注意：zipWithIndex会严格保留原RDD的行顺序，所以要确保原DataFrame的行顺序和评分列表完全对应。

额外提醒：为啥不推荐用UDF的方式？ #

除了分布式环境的问题，还有两个核心原因：

• UDF的性能远不如Spark内置函数，数据量越大差异越明显
• 全局列表的修改是线程不安全的，在多线程的executor环境中会导致不可预测的结果

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