爲什麼foreach操作符在使用Dataset時會失敗並出現NullPointerException？

Question

這是我的函數：

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得到Dataframes？

Answer 1

foreach對執行者執行操作，而dataFrame僅在驅動程序上可用。

您應該在foreach之前collect。隨着變化，foreach然後是斯卡拉的不火花的foreach。

Answer 2

您不能在轉換或操作中使用dataFrame。您需要先收集您的用戶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 
}