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

下面我提供我的代码。我遍历DataFrame&＃xff0c;prodRows并为每个product_PK找到了匹配的product_PKs子列表prodRows。

numRecProducts &＃61; 10

var listOfProducts: Map[Long,Array[(Long, Int)]] &＃61; Map()

prodRows.foreach{ row : Row &＃61;>

val product_PK &＃61; row.get(row.fieldIndex("product_PK")).toString.toLong

val gender &＃61; row.get(row.fieldIndex("gender_PK")).toString

val selection &＃61; prodRows.filter($"gender_PK" &＃61;&＃61;&＃61; gender || $"gender_PK" &＃61;&＃61;&＃61; "UNISEX").limit(numRecProducts).select($"product_PK")

var productList: Array[(Long, Int)] &＃61; Array()

if (!selection.rdd.isEmpty()) {

productList &＃61; selection.rdd.map(x &＃61;> (x(0).toString.toLong,1)).collect()

}

listOfProducts &＃61; listOfProducts &＃43; (product_PK -> productList)

}

但是当我执行它时&＃xff0c;它给了我以下错误。selection在某些迭代中看起来像是空的。但是&＃xff0c;我不明白如何处理此错误&＃xff1a;

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)

这是什么意思&＃xff0c;我该如何处理&＃xff1f;