我正在开发一个名为PySpark Cassandra的Scala/Python库。在其中,我必须在保存数据时以 pickle 格式处理序列化的 Python 对象。
我有一个工作因堆栈失败:
org.apache.spark.SparkException: Job aborted due to stage failure: Task 250 in stage 2.0 failed 4 times, most recent failure: Lost task 250.3 in stage 2.0 (TID 411, sp-prod-adg02.priv.tgho.nl): java.lang.StackOverflowError
at pyspark_cassandra.UnpickledUUIDConverter$$anonfun$convertPF$1.applyOrElse(Pickling.scala:121)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:165)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:166)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:166)
...
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:166)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:166)
启动此跟踪的代码是:
object UnpickledUUIDConverter extends TypeConverter[UUID] {
val tt = typeTag[UUID]
def targetTypeTag = tt
def convertPF = { case holder: UUIDHolder => holder.uuid }
}
位于 https://github.com/TargetHolding/pyspark-cassandra/blob/master/src/main/scala/pyspark_cassandra/Pickling.scala#L118(有关更多详细信息和上下文)。
UUIDHolder 类定义为:
class UUIDHolder {
var uuid: UUID = null
def __setstate__(values: HashMap[String, Object]): UUID = {
val i = values.get("int").asInstanceOf[BigInteger]
val buffer = ByteBuffer.wrap(i.toByteArray())
uuid = new UUID(buffer.getLong(), buffer.getLong())
uuid
}
}
(这个类的奇怪结构是为了与py4j兼容以及UUID对象如何被Python腌制)
但是我对 Scala 以及案例块和 Partial Functions 之间的关系的理解相当有限。尤其是我的案例块与 https://github.com/scala/scala/blob/2.10.x/src/library/scala/PartialFunction.scala#L166 的关系(我在 Scala 2.10.5 上运行)
使我的情况恶化:)我很难始终如一地重现错误。它发生在不同节点上的 Spark 作业中,但不是所有时间。我有一个数据集,保存该数据集时存在问题。但我无法将其固定到数据集中的特定记录。
无论如何,我都不希望使用此代码出现堆栈溢出。任何帮助将不胜感激!
回答简单的问题:
您的案例块是一个部分函数文字,如此处所述,又名模式匹配匿名函数。这是有效的,因为convertPF
的返回类型是分部函数。
它获得一个如此处所述的applyOrElse
,避免调用ifDefined
然后apply
。
堆栈上的OrElse
正是包装pf1 orElse pf2
。它实现的applyOrElse
每个部分函数的委托。
很长的pfi orElse pfi_++
链可能会在评估或orElse_i orElse (orElse_i++ orElse ...)
时溢出堆栈。