为什么使用 Dataset 时 foreach 运算符会因 NullPointerException 而失败？答案

【问题标题】：Why does foreach operator fail with NullPointerException when using Dataset?为什么使用 Dataset 时 foreach 运算符会因 NullPointerException 而失败？
【发布时间】：2017-09-18 14:50:58
【问题描述】：

这是我的功能：

def TestForeach(dataFrame: DataFrame)={
  dataFrame.select("user_id").dropDuplicates().foreach(row =>{
  dataFrame.filter("user_id == "+row.getString(0))
  })
}

我收到此错误：

ERROR Executor: Exception in task 2.0 in stage 4.0 (TID 16)
java.lang.NullPointerException
at org.apache.spark.sql.Dataset.filter(Dataset.scala:1318)

如何获取具有相同 user_id 的 Dataframe？

【问题讨论】：

能否请您包含问题的例外情况？这样会更完整。谢谢！

标签： scala apache-spark apache-spark-sql

【解决方案1】：

foreach 在执行器上执行操作，而dataFrame 仅在驱动程序上可用。

你应该在foreach 之前collect。随着变化，foreach 将不再是 Scala 的 foreach。

【讨论】：

【解决方案2】：

您不能在转换或操作中使用数据帧。您需要先收集您的用户 ID：

def testForeach(dataFrame: DataFrame): Seq[DataFrame] = {
  val userIds: Array[String] = dataFrame.select("user_id").distinct.map(_.getString(0)).collect
  userIds.map(uid => dataFrame.filter($"user_id" === uid)).toSeq
}

【讨论】：