Spark DataFrame行转列后按课程拆分CSV：仅保留对应课程课时列

问题的核心在于你之前的方法是先对全量数据做pivot，生成包含所有课程课时列的DataFrame，再用partitionBy分区存储——这只会按course_id拆分数据行，但DataFrame的Schema还是包含所有列，所以每个CSV里会出现其他课程的列（值为null）。

要实现每个CSV仅保留对应课程的专属列，我们需要按course_id拆分DataFrame，对每个课程单独做行转列后再保存，具体步骤如下：

1. 获取所有唯一的课程ID #

首先提取数据中所有不重复的course_id，方便后续逐个处理：

import org.apache.spark.sql.functions._

val someDF = Seq(
  ("user1", "math","algebra-1","90"),
  ("user1", "physics","gravity","70"),
  ("user3", "biology","health","50"),
  ("user2", "biology","health","100"),
  ("user1", "math","algebra-1","40"),
  ("user2", "physics","gravity-2","20")
).toDF("user_id", "course_id","lesson_name","score")

// 获取所有唯一的course_id
val courseIds = someDF.select("course_id").distinct().collect().map(_.getString(0))

2. 逐个处理每个课程并保存 #

遍历每个course_id，过滤出对应课程的数据，单独做行转列，这样生成的DataFrame只会包含该课程的课时列，最后保存为CSV：

val somepath = "/your/save/path" // 替换为你的实际保存路径

courseIds.foreach { courseId =>
  // 过滤当前课程的数据
  val courseSpecificDF = someDF.filter($"course_id" === courseId)
  
  // 对当前课程数据做行转列，仅生成该课程的课时列
  val pivotedDF = courseSpecificDF
    .groupBy("user_id", "course_id")
    .pivot("lesson_name")
    .agg(first("score")) // 这里可根据需求替换聚合函数，比如sum/avg等
  
  // 保存为单独的CSV文件
  pivotedDF.write
    .mode("overwrite")
    .format("com.databricks.spark.csv")
    .option("header", "true")
    .save(s"$somepath/$courseId.csv")
}

效果验证 #

• biology.csv 的列：user_id, course_id, health
• physics.csv 的列：user_id, course_id, gravity, gravity-2
• math.csv 的列：user_id, course_id, algebra-1

完全符合你期望的输出，每个CSV仅包含对应课程的专属列，不会出现无关列。

注意：如果你的course_id数量非常多（比如上万级），collect() 把所有course_id拉到Driver可能会有内存压力，这种情况下可以考虑用foreachPartition或者Spark SQL的方式批量处理，但对于大多数场景，上面的方法已经足够高效。

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