调用 udf 触发数据帧时任务不可序列化错误答案

【问题标题】：Task not serializable error while calling udf to spark dataframe调用 udf 触发数据帧时任务不可序列化错误
【发布时间】：2020-05-25 03:10:04
【问题描述】：

我有一个用于加密的 scala 函数，然后从中创建了一个 udf 并将其传递给我的 als_embeddings 数据框中的列之一，以将新列添加到我的数据框中。

import java.util.Base64
import javax.crypto.Cipher
import javax.crypto.spec.{IvParameterSpec, SecretKeySpec}

val Algorithm = "AES/CBC/PKCS5Padding"
val Key = new SecretKeySpec(Base64.getDecoder.decode("BiwHeIqzQa8X6MXtdg/hhQ=="), "AES")
val IvSpec = new IvParameterSpec(new Array[Byte](16))

def encrypt(text: String): String = {
  val cipher = Cipher.getInstance(Algorithm)
  cipher.init(Cipher.ENCRYPT_MODE, Key, IvSpec)

  new String(Base64.getEncoder.encode(cipher.doFinal(text.getBytes("utf-8"))), "utf-8")
}


val encryptUDF = udf((uid : String) => encrypt(uid))

将encryptUDF 以上传递给我的 spark 数据框以创建一个带有加密 uid 的新列

val als_encrypt_embeddings = als_embeddings.withColumn("encrypt_uid",encryptUDF(col("uid")))
als_encrypt_embeddings.show()

但是当我这样做时，它给了我以下错误：

线程“主”org.apache.spark.SparkException 中的异常：任务不是可序列化

我在这里错过了什么。

【问题讨论】：

看看stackoverflow.com/questions/22592811/…
由于您已经将 encryptUDF 声明为 val，因此请尝试将函数实现本身放在那里，而不是创建 def 然后将其分配给 value 函数。如果它有效，那么它为什么有效，您可以参考上面分享的另一篇 stackoverflow 帖子。

标签： scala apache-spark encryption user-defined-functions

【解决方案1】：

错误信息Task not serializable是正确的，但不是很清楚。在堆栈跟踪的更下方，有更详细的解释出了什么问题：

Exception in thread "main" org.apache.spark.SparkException: Task not serializable
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:403)
    at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:393)
    at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:162)
[...]
Caused by: java.io.NotSerializableException: javax.crypto.spec.IvParameterSpec
Serialization stack:
    - object not serializable (class: javax.crypto.spec.IvParameterSpec, value: javax.crypto.spec.IvParameterSpec@7d4d65f5)
    - field (class: Starter$$anonfun$1, name: IvSpec$1, type: class javax.crypto.spec.IvParameterSpec)
    - object (class Starter$$anonfun$1, <function1>)
    - element of array (index: 2)
    - array (class [Ljava.lang.Object;, size 3)
    - field (class: org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13, name: references$1, type: class [Ljava.lang.Object;)
    - object (class org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13, <function2>)
    at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
    at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
    at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:400)
    ... 48 more

在堆栈跟踪的Caused by 部分，Spark 报告它无法序列化javax.crypto.spec.IvParameterSpec 的实例。

ParameterSpec 已在驱动程序 JVM 中创建，同时 udf 在其中一个执行程序上执行。因此，必须对对象进行序列化才能将其移动到执行程序的 VM。由于对象不可序列化，因此移动它的尝试失败。

解决问题的最简单方法是通过将代码块移动到 udf 的闭包中，直接在执行程序的 VM 中创建加密所需的对象：

val encryptUDF = udf((uid : String) => {
  val Algorithm = "AES/CBC/PKCS5Padding"
  val Key = new SecretKeySpec(Base64.getDecoder.decode("BiwHeIqzQa8X6MXtdg/hhQ=="), "AES")
  val IvSpec = new IvParameterSpec(new Array[Byte](16))

  def encrypt(text: String): String = {
    val cipher = Cipher.getInstance(Algorithm)
    cipher.init(Cipher.ENCRYPT_MODE, Key, IvSpec)

    new String(Base64.getEncoder.encode(cipher.doFinal(text.getBytes("utf-8"))), "utf-8")
  }
  encrypt(uid)
})

这样所有对象都将直接在 executors 虚拟机中创建。

这种方法的缺点是每次调用 udf 都会创建一组加密对象。如果这些对象的实例化成本很高，这可能会导致性能问题。一种选择是使用mapPartitions 而不是udf。在此answer mapPartitions 用于避免在迭代数据帧时创建太多昂贵的数据库连接。这种方法也可以用在这里。

【讨论】：

【解决方案2】：

我们可以将函数定义为不引用不可序列化值的独立对象的一部分。

object EncryptUtils extends Serializable {
  ...
  ...
  ...
}

【讨论】：