Искра и javax.crypto

Question

Я пытаюсь узнать о шифровании, и я нашел следующий код

import java.util.Base64
object JavaCryptoEncryption{

  val Algorithm = "AES/CBC/PKCS5Padding"
  val IvSpec = new IvParameterSpec(new Array[Byte](16))

  def encrypt(text: String, b64secret: String): String = {
    val cipher = Cipher.getInstance(Algorithm)
    val key = new SecretKeySpec(Base64.getDecoder.decode(b64secret), "AES")
    cipher.init(Cipher.ENCRYPT_MODE, key, IvSpec)

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

  def decrypt(text: String, b64secret: String): String = {
    val cipher = Cipher.getInstance(Algorithm)
    val key = new SecretKeySpec(Base64.getDecoder.decode(b64secret), "AES")
    cipher.init(Cipher.DECRYPT_MODE, key, IvSpec)

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

В другом месте в моей системе я определяю и храню ключ. Затем я применяю JavaCryptoEncryption.encrypt и JavaCryptoEncryption.decrypt к строковому значению, и оно отлично работает. Однако, когда я хочу преобразовать их в UDF и применить к столбцу DataFrame, я получаю org.apache.spark.SparkException: Task not serializable. Аналогичный код (без iv) работает для AES / ECB / PKCS5Padding. Некоторые режимы не поддерживают параллелизм? Есть ли способ обойти? Или, может быть, есть другая причина?

Answer 1

Мне кажется, проблема в том, что вы создаете экземпляр IvParameterSpec в драйвере, а затем Spark пытается его сериализовать и отправить исполнителям (где фактически выполняются все пользовательские функции). Может быть, попробуйте поместить весь код, связанный с созданием объектов, в сам UDF и просто указать свой ключ в качестве параметра UDF? Таким образом, вам нужно будет отправить только String рабочим, поэтому не должно быть проблем с сериализацией.