如何在PySpark中将多列单元格值拆分后展开为对应多行

你需要的功能可以通过arrays_zip配合单次explode实现，核心逻辑是先把所有列拆分后的数组合并为同位置元素绑定的结构体数组，再一次性展开，完全避免笛卡尔积和高内存占用的join操作。

可运行代码示例 #

from pyspark.sql import functions as F

# 1. 先将所有需要展开的列按分隔符拆分为数组
df = df \
    .withColumn("arr1", F.split(F.col("col1"), ",")) \
    .withColumn("arr2", F.split(F.col("col2"), ",")) \
    .withColumn("arr3", F.split(F.col("col3"), ","))

# 2. 将多个数组合并为结构体数组，同下标元素绑定为一个结构体
df = df.withColumn("zipped_arr", F.arrays_zip(F.col("arr1"), F.col("arr2"), F.col("arr3")))

# 3. 单次展开结构体数组，不会产生笛卡尔积
df = df.withColumn("zipped", F.explode(F.col("zipped_arr")))

# 4. 从展开后的结构体中提取各列，保留原ID
df = df.select(
    F.col("ID"),
    F.col("zipped.arr1").alias("col1"),
    F.col("zipped.arr2").alias("col2"),
    F.col("zipped.arr3").alias("col3")
)

如果需要简化代码，也可以合并为链式写法：

from pyspark.sql import functions as F

df = df.select(
    "ID",
    F.explode(F.arrays_zip(
        F.split("col1", ",").alias("col1"),
        F.split("col2", ",").alias("col2"),
        F.split("col3", ",").alias("col3")
    )).alias("res")
).select("ID", "res.col1", "res.col2", "res.col3")

原方法不符合预期/报错的原因 #

• 分开对每列执行explode：每次explode都会将当前所有行和新数组的所有元素做笛卡尔积，三次操作后行数就是三个数组长度的乘积，和需求不符。
• 单独拆分后join：同ID下的行关联会产生大量shuffle，内存占用高，大数据量下无法正常运行。
• stack函数报错：stack是行转列专用函数，需要先指定展开的行数，再传入对应列值，你的用法不符合函数参数要求，所以报错。

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