编写PySpark代码实现指定数据转换（禁用stack函数）

输入数据结构 #

假设输入数据包含以下字段：

• id: 学生ID
• Maths_Score: 数学分数
• Maths_Grade: 数学等级
• Science_Score: 科学分数
• Science_Grade: 科学等级
• English_Score: 英语分数
• English_Grade: 英语等级

目标输出结构 #

转换后的数据需包含以下字段：

• id: 学生ID
• Subject: 科目名称
• Score: 对应科目分数
• Grade: 对应科目等级

实现代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit

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

# 模拟输入数据
input_data = [
    (1, 85, "A", 90, "A+", 78, "B"),
    (2, 70, "B", 82, "A", 88, "A")
]
input_df = spark.createDataFrame(
    input_data,
    ["id", "Maths_Score", "Maths_Grade", "Science_Score", "Science_Grade", "English_Score", "English_Grade"]
)

# 处理数学科目数据
maths_df = input_df.select(
    "id",
    input_df["Maths_Score"].alias("Score"),
    input_df["Maths_Grade"].alias("Grade")
).withColumn("Subject", lit("Maths"))

# 处理科学科目数据
science_df = input_df.select(
    "id",
    input_df["Science_Score"].alias("Score"),
    input_df["Science_Grade"].alias("Grade")
).withColumn("Subject", lit("Science"))

# 处理英语科目数据
english_df = input_df.select(
    "id",
    input_df["English_Score"].alias("Score"),
    input_df["English_Grade"].alias("Grade")
).withColumn("Subject", lit("English"))

# 合并所有科目数据
output_df = maths_df.unionByName(science_df).unionByName(english_df)

# 展示结果
output_df.orderBy("id", "Subject").show()

代码说明 #

• 分别提取每个科目的分数、等级字段，并重命名为统一的Score和Grade，同时添加固定的Subject标识
• 使用unionByName确保合并时列名匹配，避免因列顺序问题导致错误
• 最后按id和Subject排序，使结果更规整

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