由：org.apache.spark.sql.Dataset上的java.lang

Question

下面我提供我的代码。我遍历DataFrame，prodRows并为每个product_PK找到了匹配的product_PKs子列表prodRows。  numRecProducts = 10  var listOfProducts: Map[Long,Array[(Long, Int)]] = Map()  prodRows.foreach{ row : Row =>      val product_PK = row.get(row.fieldIndex("product_PK")).toString.toLong      val gender = row.get(row.fieldIndex("gender_PK")).toString      val selection = prodRows.filter($"gender_PK" === gender || $"gender_PK" === "UNISEX").limit(numRecProducts).select($"product_PK")      var productList: Array[(Long, Int)] = Array()      if (!selection.rdd.isEmpty()) {        productList = selection.rdd.map(x => (x(0).toString.toLong,1)).collect()      }    listOfProducts = listOfProducts + (product_PK -> productList)  }但是当我执行它时，它给了我以下错误。selection在某些迭代中看起来像是空的。但是，我不明白如何处理此错误：Driver stacktrace:    at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1690)    at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1678)    at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1677)    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)    at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1677)    at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:855)    at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:855)    at scala.Option.foreach(Option.scala:257)    at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:855)    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1905)这是什么意思，我该如何处理？

Answer 1

问题是您尝试prodRows从内部访问prodRows.foreach。您不能在转换中使用数据框，数据框仅存在于驱动程序中。