为 Spark UDF 执行提供上下文

Question

我正在使用 Scala 编程语言。我想用 sha2 和 salt 对整个数据框列进行哈希处理。我已经实现了以下 UDF，它应该采用 MessageDigest 和将被散列的输入字符串。

  val md = MessageDigest.getInstance("SHA-256")

  val random = new SecureRandom();
  val salt: Array[Byte] = new Array[Byte](16)
  random.nextBytes(salt)
  md.update(salt)

  dataFrame.withColumn("ColumnName", Sqlfunc(md, col("ColumnName")))

  ....some other code....

  val HashValue: ((MessageDigest, String) => String) = (md: MessageDigest, input: String) =>
  {
    val hashedPassword: Array[Byte] = md.digest(input.getBytes(StandardCharsets.UTF_8))
    val sb: StringBuilder = new StringBuilder
    for (b <- hashedPassword) {sb.append(String.format("%02x", Byte.box(b)))}
    sb.toString();
  }

  val Sqlfunc = udf(HashValue)

但是上面的代码无法编译，因为我不知道如何将 messageDigest 传递给这个函数，所以我遇到了以下错误

 <<< ERROR!
java.lang.ClassCastException: com...................$$anonfun$9 cannot be cast to scala.Function1

谁能告诉我我做错了什么？ 另外，我是密码学的新手，所以请随意提出任何建议。我们必须使用 Sha2 和盐。 你觉得这里的表现怎么样？

谢谢

Answer 1

MessageDigest 不在您的数据中。这只是 UDF 评估的上下文。这种类型的上下文是通过closures 提供的。

有很多方法可以达到预期的效果。以下是使用函数currying 的有用模式：

object X extends Serializable {
  import org.apache.spark.sql.expressions.UserDefinedFunction
  import org.apache.spark.sql.functions.udf

  def foo(context: String)(arg1: Int, arg2: Int): String =
    context.slice(arg1, arg2)

  def udfFoo(context: String): UserDefinedFunction =
    udf(foo(context) _)
}

试一试：

spark.range(1).toDF
  .select(X.udfFoo("Hello, there!")('id, 'id + 5))
  .show(false)

生成

+-----------------+
|UDF(id, (id + 5))|
+-----------------+
|Hello            |
+-----------------+