PySpark DataFrame按列值复制行及代码中获取No_of_Occ的问题

嘿，我看了你写的duplicate_function，发现几个小细节问题，先帮你修正过来，再一步步说明怎么正确应用它来实现按列值复制行的需求：

修正后的函数代码 #

首先，你的函数里有个变量名写错了（to_return应该是data），还有类型转换可以更贴合实际需求，修正后的代码如下：

from pyspark.sql import Row

def duplicate_function(row):
    data = []  # 存储要返回的所有新行
    # 复制次数应该是整数，所以这里转成int更合理（如果No_of_Occ是字符串类型的话）
    to_duplicate = int(row["No_of_Occ"])
    for i in range(to_duplicate):
        row_dict = row.asDict()  # 把Spark Row转成Python字典方便修改
        row_dict["No_of_Occ"] = str(i)  # 更新No_of_Occ为当前循环的索引值
        new_row = Row(**row_dict)  # 重新构建Spark Row对象
        data.append(new_row)  # 将新行加入列表（之前这里误用了未定义的to_return）
    return data

我把while循环换成了更简洁的for循环，逻辑是一样的，看起来更清爽。

如何将函数应用到DataFrame #

PySpark的普通UDF没办法直接返回多行数据，所以我们需要借助RDD的flatMap方法——它可以把每个原始行生成的多行结果展开，最后再转回DataFrame：

假设你的原始DataFrame叫df，代码示例如下：

# 转换为RDD处理，再转回DataFrame（保持原schema）
result_df = df.rdd.flatMap(duplicate_function).toDF(df.schema)

# 查看最终结果
result_df.show()

这种方法会严格保持原始数据的顺序，完全符合你的需求。

另一种更优雅的DataFrame API写法 #

如果你不想用RDD，也可以用PySpark内置的函数来实现，不需要自定义函数，性能可能还更好：

from pyspark.sql import functions as F

result_df = df.withColumn(
    "temp_index",
    # 生成从0到No_of_Occ-1的序列，再炸开成多行
    F.explode(F.sequence(F.lit(0), F.col("No_of_Occ") - 1))
).withColumn(
    "No_of_Occ",
    F.col("temp_index").cast("string")  # 更新No_of_Occ的值
).drop("temp_index")  # 删掉临时列

这种写法更贴合PySpark的DataFrame风格，不需要手动处理Row对象，代码更简洁。

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