如何用PySpark或SQL将1行4列DataFrame转为4行2列？

场景说明 #

假设你的原始DataFrame是多列结构，需要转换为2列N行的结构（N等于原列数）：一列存储原列名（category），另一列存储对应的值（value）。以下提供两种动态实现方式，无需硬编码列名。

方法1：使用stack函数动态生成表达式 #

stack是Spark SQL内置的unpivot函数，适合快速实现列转行。通过动态拼接表达式可适配任意列数的DataFrame：

from pyspark.sql import SparkSession
from pyspark.sql.functions import expr

# 初始化Spark会话
spark = SparkSession.builder.appName("DynamicUnpivot").getOrCreate()

# 模拟你的原始DataFrame（替换成实际数据）
sample_data = [("val1", "val2", "val3", "val4")]
sample_cols = ["A_col", "B_col", "C_col", "D_col"]
original_df = spark.createDataFrame(sample_data, sample_cols)

# 动态生成stack表达式：stack(列数, 列名1, 列值1, 列名2, 列值2...)
col_count = len(original_df.columns)
stack_expression = f"stack({col_count}, {', '.join([f'{repr(col)}, {col}' for col in original_df.columns])}) as (category, value)"

# 执行转换
unpivoted_df = original_df.selectExpr(stack_expression)

# 查看结果
unpivoted_df.show()

执行后输出：

+---------+-----+
| category|value|
+---------+-----+
|    A_col| val1|
|    B_col| val2|
|    C_col| val3|
|    D_col| val4|
+---------+-----+

方法2：自定义melt函数（类似Pandas） #

如果需要更灵活的控制（比如保留分组ID列），可以实现类似Pandas的melt函数：

from pyspark.sql import SparkSession
from pyspark.sql.functions import create_map, explode, lit

spark = SparkSession.builder.appName("DynamicUnpivot").getOrCreate()

# 模拟原始DataFrame
sample_data = [("val1", "val2", "val3", "val4")]
sample_cols = ["A_col", "B_col", "C_col", "D_col"]
original_df = spark.createDataFrame(sample_data, sample_cols)

def melt(df, id_columns=[], value_columns=None, var_col_name="category", value_col_name="value"):
    # 默认转换所有非ID列
    if not value_columns:
        value_columns = [col for col in df.columns if col not in id_columns]
    
    # 创建列名与列值的映射，再展开映射
    column_map = create_map(*[lit(col), df[col] for col in value_columns])
    return df.select(*id_columns, explode(column_map).alias(var_col_name, value_col_name))

# 执行转换（无ID列时直接调用）
unpivoted_df = melt(original_df)

unpivoted_df.show()

扩展：保留分组ID列

如果你的原始DataFrame包含分组列（比如ID），只需传入id_columns参数：

# 模拟带ID列的原始DataFrame
grouped_data = [(1, "val1", "val2", "val3", "val4"), (2, "val5", "val6", "val7", "val8")]
grouped_cols = ["ID", "A_col", "B_col", "C_col", "D_col"]
grouped_df = spark.createDataFrame(grouped_data, grouped_cols)

# 保留ID列的转换
unpivoted_grouped_df = melt(grouped_df, id_columns=["ID"])
unpivoted_grouped_df.show()

输出会保留ID列：

+---+---------+-----+
| ID| category|value|
+---+---------+-----+
|  1|    A_col| val1|
|  1|    B_col| val2|
|  1|    C_col| val3|
|  1|    D_col| val4|
|  2|    A_col| val5|
|  2|    B_col| val6|
|  2|    C_col| val7|
|  2|    D_col| val8|
+---+---------+-----+

内容的提问来源于stack exchange，提问作者Ramaraju.d