无法对DF执行DataFrame操作求助：Scala中Spark SQL查询结果处理问题

嘿，我一眼就揪出问题所在啦！你调用了collect()方法，这直接把Spark的DataFrame转换成了本地的Array[org.apache.spark.sql.Row]数组——这就是你没法再执行DataFrame相关操作的核心原因！

为什么会这样？ #

collect()的作用是把分布式存储在集群节点上的DataFrame数据，全部拉取到你的驱动程序所在的本地机器里，转换成普通的Scala数组。数组是本地集合，自然不支持DataFrame特有的API（比如filter()、groupBy()、join()这些）。

解决方案分两种情况： #

情况1：还没执行collect()，从头调整

直接去掉collect()，保留DataFrame对象就好：

// 这样dF的类型就是DataFrame，能正常使用所有DataFrame操作
val dF = sqlContext.sql("select * from employeeTable")

比如你想筛选年龄大于25的员工，就可以直接写：

val filteredDF = dF.filter($"age" > 25)

情况2：已经执行了collect()，想转回DataFrame

如果已经得到了Array[Row]，可以通过Spark的createDataFrame方法把它转回DataFrame，但需要手动指定数据的schema（因为数组本身不保存结构信息）：

import org.apache.spark.sql.types._

// 先定义和原表一致的schema
val employeeSchema = StructType(Seq(
  StructField("id", IntegerType, nullable = false),
  StructField("name", StringType, nullable = false),
  StructField("age", IntegerType, nullable = false),
  StructField("gender", StringType, nullable = false),
  StructField("level", IntegerType, nullable = false),
  StructField("salary", IntegerType, nullable = false)
))

// 假设你已经有了从collect()得到的数组
val arrayRows = sqlContext.sql("select * from employeeTable").collect()
// 把数组转成RDD，再结合schema创建DataFrame
val restoredDF = sqlContext.createDataFrame(sc.parallelize(arrayRows), employeeSchema)

这样restoredDF就又是DataFrame类型，能正常操作了。

小提醒 #

collect()一定要谨慎使用！只有当你确定数据集非常小（比如测试数据）的时候才用，因为它会把整个数据集拉到本地，数据量大的话直接会把驱动节点内存撑爆，导致程序崩溃。日常处理大数据时，尽量用DataFrame/Dataset的分布式API操作，别轻易拉取本地。

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