PySpark如何高效实现列转行且不损失其并行计算能力

直接使用PySpark原生的arrays_zip + explode算子即可实现需求，全程分布式运行，不会损失并行计算能力，性能远高于多次union的写法。

实现代码 #

首先导入依赖函数：

from pyspark.sql import functions as F

假设你的原始DataFrame名为raw_df，按如下逻辑处理：

# 1. 按对应关系配置ID列和Name列列表
id_col_list = ["ID_1", "ID_2", "ID_3", "ID_4"]
name_col_list = ["Name1", "Name2", "Name3", "Name4"]

# 2. 核心转换逻辑
result_df = raw_df.select(
    "USER",
    # 打包对应位置的ID和Name为数组后炸开，每行对应一组ID/Name
    F.explode(F.arrays_zip(
        F.array(*id_col_list),
        F.array(*name_col_list)
    )).alias("temp_col")
).select(
    "USER",
    F.col("temp_col.0").alias("ID"),
    F.col("temp_col.1").alias("Name")
# 3. 过滤掉空值的无效行
).filter(
    F.col("ID").isNotNull() & F.col("Name").isNotNull()
)

执行result_df.show()即可得到你期望的输出结果。

方案优势 #

• 完全基于PySpark原生分布式算子实现，无自定义UDF，无Driver端数据拉取操作，100%保留集群并行计算能力
• 仅对原始表做1次全量扫描，不需要像union方案那样每多一组ID/Name就多扫一次表，数据量越大、列数越多，性能优势越明显
• 拓展性强，后续新增列组仅需要修改id_col_list和name_col_list的配置即可，不需要调整核心转换逻辑

补充说明（不推荐的union方案） #

你也可以通过多次union的方式实现，但性能更低，仅适合列数极少的场景：

# 仅做参考，不推荐用于生产环境
part1 = raw_df.select("USER", F.col("ID_1").alias("ID"), F.col("Name1").alias("Name")).filter("ID IS NOT NULL")
part2 = raw_df.select("USER", F.col("ID_2").alias("ID"), F.col("Name2").alias("Name")).filter("ID IS NOT NULL")
part3 = raw_df.select("USER", F.col("ID_3").alias("ID"), F.col("Name3").alias("Name")).filter("ID IS NOT NULL")
part4 = raw_df.select("USER", F.col("ID_4").alias("ID"), F.col("Name4").alias("Name")).filter("ID IS NOT NULL")
result_df = part1.unionByName(part2).unionByName(part3).unionByName(part4)

内容的提问来源于stack exchange，提问作者Carlos Eduardo Bilar Rodrigues