时间戳

首页 > 解决方案 > 调用 udf 触发数据帧时任务不可序列化错误

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

问题描述

我有一个用于加密的 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 中的异常：任务不可序列化

我在这里想念什么。

标签： scalaapache-sparkencryptionuser-defined-functions

解决方案

错误信息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 bystacktrace 的一部分中，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 VM 中创建。

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

推荐阅读