如何基于列值拆分PySpark DataFrame并转换为学生信息宽表

实现思路 #

• 第一步：给原始数据添加单调递增的行索引，确保行顺序和原始输入一致（Spark DataFrame默认无固定行序，必须先显式指定顺序才能正确按Roll no分段）
• 第二步：生成学生分组ID：遇到Roll no行就给分组计数+1，同一学生的所有属性行会被分到同一个ID下
• 第三步：按分组ID聚合，行转列（pivot）得到最终的结构化宽表

完整代码 #

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 初始化SparkSession
spark = SparkSession.builder.appName("StudentPivot").getOrCreate()

# ------------------- 示例数据构造（实际使用时替换为你自己的数据源读取代码） -------------------
data = [
    ("Roll no", "141641"),
    ("Name", "SUNNY"),
    ("Mobile", "4567892345"),
    ("Roll no", "141633"),
    ("Name", "Vikas"),
    ("Email", "V.com")
]
df = spark.createDataFrame(data, schema=["Column", "Value"])
# ------------------------------------------------------------------------------------------

# 1. 添加行索引保证顺序
df_with_idx = df.withColumn("row_idx", F.monotonically_increasing_id())

# 2. 生成分组ID
window_spec = Window.orderBy("row_idx")
# 遇到Roll no就标记为1，累加后就是分组ID
df_with_group = df_with_idx.withColumn(
    "is_rollno",
    F.when(F.col("Column") == "Roll no", 1).otherwise(0)
).withColumn(
    "group_id",
    F.sum("is_rollno").over(window_spec)
)

# 3. 行转列得到宽表
result_df = df_with_group.groupBy("group_id").pivot("Column").agg(F.first("Value"))

# 可选：调整列顺序、重命名、删除不需要的列
result_df = result_df.select(
    F.col("Roll no").alias("Roll No"),
    "Name",
    "Mobile",
    "Email"
)

# 输出结果
result_df.show()

代码说明 #

• monotonically_increasing_id()生成的ID是按分区自增的，整体顺序和DataFrame的扫描顺序一致，满足保留原始行序的需求
• pivot操作会自动识别所有动态的属性列，后续如果出现Address、Class等新属性也会自动生成对应列，不需要手动修改代码
• 缺失的属性值会自动填充为null，和要求的输出格式一致

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