我们正在将spark作业从2.4.5迁移到3.1.1(scala 2.11.11到scala 2.12.13)
我们面临以下例外情况
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:416) ~[spark-core_2.12-3.1.1.jar:3.1.1]
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:406) ~[spark-core_2.12-3.1.1.jar:3.1.1]
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:162) ~[spark-core_2.12-3.1.1.jar:3.1.1]
at org.apache.spark.SparkContext.clean(SparkContext.scala:2465) ~[spark-core_2.12-3.1.1.jar:3.1.1]
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$1(RDD.scala:912) ~[spark-core_2.12-3.1.1.jar:3.1.1]
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) ~[spark-core_2.12-3.1.1.jar:3.1.1]
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) ~[spark-core_2.12-3.1.1.jar:3.1.1]
at org.apache.spark.rdd.RDD.withScope(RDD.scala:414) ~[spark-core_2.12-3.1.1.jar:3.1.1]
at org.apache.spark.rdd.RDD.mapPartitionsWithIndex(RDD.scala:911) ~[spark-core_2.12-3.1.1.jar:3.1.1]
at org.apache.spark.sql.execution.WholeStageCodegenExec.doExecute(WholeStageCodegenExec.scala:749) ~[spark-sql_2.12-3.1.1.jar:3.1.1]
当我们在数据集上使用map,flatmap,cogroup(带有一些函数)时,就会触发这种情况。例如,下面是我们用来将一个数据集Map到另一个数据集的函数的签名:
def myFunction(var: (Seq[caseClass1], Option[caseClass2], Option[caseClass3])): (Seq[caseClass2], Seq[caseClass3], Seq[caseClass2], Seq[caseClass3])
代码没有改变。我们使用kryoserializer,配置几乎相同。
它曾经在spark 2.4.5中工作过。
我们找到了一个可以让我工作的地方。我们将其 Package 为以下函数:
def genMapper[A, B](f: A => B): A => B = {
val locker = com.twitter.chill.MeatLocker(f)
x => locker.get.apply(x)
}
如下
DS.map(genMapper(myFunction)) // This works
// DS.map(myFunction) -- This does not work
然而,我真的很想了解spark 2.4.5之间的序列化行为有什么根本的不同。和spark 3.x。。
谢谢您
暂无答案!
目前还没有任何答案,快来回答吧!