如何从Scala方法创建UDF(计算md5)？

Question

我想从两个已经工作的函数构建一个UDF.我正在尝试将md5哈希计算为现有Spark Dataframe的新列.

def md5(s: String): String = { toHex(MessageDigest.getInstance("MD5").digest(s.getBytes("UTF-8")))}
def toHex(bytes: Array[Byte]): String = bytes.map("%02x".format(_)).mkString("")

结构(到目前为止我有什么)

val md5_hash: // UDF Implementation
val sqlfunc = udf(md5_hash)
val new_df = load_df.withColumn("New_MD5_Column", sqlfunc(col("Duration")))

不幸的是,我不知道如何将该函数作为UDF实现.

Answer 1

为什么不使用内置的md5功能？

md5(e:Column):Column计算二进制列的MD5摘要,并将该值作为32个字符的十六进制字符串返回.

然后您可以按如下方式使用它:

val new_df = load_df.withColumn("New_MD5_Column", md5($"Duration"))

您必须确保该列是二进制类型,因此如果它是int,您可能会看到以下错误:

org.apache.spark.sql.AnalysisException:Duration由于数据类型不匹配,无法解析'md5()':参数1需要二进制类型,但是' Duration'是int类型.

然后,您应该md5使用bin函数将类型更改为兼容,即二进制类型.

bin(e:Column):Column一个表达式,它返回给定long列的二进制值的字符串表示形式.例如,bin("12")退货"1100".

解决方案如下:

val solution = load_df.
  withColumn("bin_duration", bin($"duration")).
  withColumn("md5", md5($"bin_duration"))
scala> solution.show(false)
+--------+------------+--------------------------------+
|Duration|bin_duration|md5                             |
+--------+------------+--------------------------------+
|1       |1           |c4ca4238a0b923820dcc509a6f75849b|
+--------+------------+--------------------------------+

您还可以将函数"链接"在一起并进行转换并将MD5计算在一个中withColumn,但我更喜欢将步骤分开,以防有问题需要解决并且中间步骤通常会有所帮助.

性能

你为什么会考虑使用内置的功能的原因bin和md5在用户自定义函数(UDF)是你能星火SQL是完全控制获得更好的性能,将不会从添加额外的步骤,序列化和反序列化内部行表示.

这不是这种情况,但仍然需要更少的导入和使用.

Answer 2

您可以使用以下udf名为的函数md5

import org.apache.spark.sql.functions._
def toHex(bytes: Array[Byte]): String = bytes.map("%02x".format(_)).mkString("")
def md5 = udf((s: String) => toHex(MessageDigest.getInstance("MD5").digest(s.getBytes("UTF-8"))))

val new_df = load_df.withColumn("New_MD5_Column", md5(col("Duration")))