任务不可序列化 - Spark 2.4 和 Spark 3.x 之间的序列化行为变化

如何解决任务不可序列化 - Spark 2.4 和 Spark 3.x 之间的序列化行为变化

我们正在将 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（带有一些函数）时触发。 例如，这里是我们用来将一个数据集映射到另一个数据集的函数的签名：

def myFunction(var: (Seq[caseClass1],Option[caseClass2],Option[caseClass3])): (Seq[caseClass2],Seq[caseClass3],Seq[caseClass2],Seq[caseClass3])

代码没有改变。我们使用KryoSerializer，配置几乎一样。

它曾经在 Spark 2.4.5 中工作。

我们找到了一种解决方法来使我的工作正常进行。我们将其包装成以下函数：

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 ..

谢谢