基于PySpark实现单行动态DataFrame到列名-值映射DataFrame的转换

问题场景 #

给定动态结构的Spark DataFrame df1（列名、列数会随加载数据变化），示例结构如下：

需要将其转换为固定结构的df2：第一列为df1的列名，第二列为对应列的取值：

静态手动构造数据的方案无法适配动态变化的df1，需要动态实现方法。

动态实现代码 #

利用PySpark的stack函数可以完美解决这个问题，该函数专门用于宽表转长表，且支持动态生成转换逻辑：

from pyspark.sql import functions as F

# 假设df1已通过数据源加载完成
# 动态生成stack表达式，适配任意列数的df1
col_count = len(df1.columns)
stack_args = ", ".join([f"{repr(col)}, {col}" for col in df1.columns])
stack_expr = f"stack({col_count}, {stack_args}) as (columnName, ColumnValues)"

# 执行转换得到df2
df2 = df1.selectExpr(stack_expr)

# 查看转换结果
df2.show()

代码说明 #

• stack函数第一个参数是需要转换的列总数，后续参数按「列名字符串、列值」的成对形式传入，最终会将每一对拆分为一行数据。
• 通过遍历df1.columns动态生成stack的参数列表，不管df1的列名、列数如何变化，都能自动适配。
• 使用repr(col)确保列名被作为字符串常量传入，避免SQL语法错误。

也可以用F.expr的方式构建表达式，效果完全一致：

stack_expr = F.expr(f"stack({len(df1.columns)}, {', '.join([f'{repr(col)}, {col}' for col in df1.columns])})")
df2 = df1.select(stack_expr.alias("columnName", "ColumnValues"))

验证效果 #

执行上述代码后，df2的输出结构和数据完全符合目标要求，且能适配任意单一行的动态宽表结构。

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