一次处理多个 Spark DataFrame 列答案

【问题标题】：Function over multiple Spark DataFrame columns at once一次处理多个 Spark DataFrame 列
【发布时间】：2017-12-15 14:22:02
【问题描述】：

我需要一次计算多个数据帧列的 md5 哈希。

功能

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

以一列为例

var test_df = load_df.as('a).select($"a.attr1", md5($"a.attr2").as("hash_key"))

+-------------+--------------------+
|     attr1   |            hash_key|
+-------------+--------------------+
|9/1/2015 0:23|7a2f516dad8f13ae1...|
|9/1/2015 0:31|339c72b1870c3a6be...|
|9/1/2015 0:19|7065847af7abc6bce...|
|9/1/2015 1:32|38c7276958809893b...|

一列 (a.attr2) 的生成效果很好，但我找不到任何将多列插入/连接到 md5() 函数中的好方法。

【问题讨论】：

标签： scala apache-spark dataframe md5 user-defined-functions

【解决方案1】：

您应该使用concat_ws，如下所示：

md5(concat_ws(",",$"a.attr2",$"a.attr3",$"a.attr4"))

这是一个例子：

Seq(("a","b","c")).toDF("x","y","z").withColumn("foo", md5(concat_ws(",",$"x",$"y",$"z"))).show(false)
// +---+---+---+--------------------------------+
// |x  |y  |z  |foo                             |
// +---+---+---+--------------------------------+
// |a  |b  |c  |a44c56c8177e32d3613988f4dba7962e|
// +---+---+---+--------------------------------+

【讨论】：

【解决方案2】：

就我个人而言，我会在 UDF 中进行连接，这会给你更多的灵活性：

例如传递字符串数组：

val md5 = udf((arrs:Seq[String]) => {
  val s = arrs.mkString(",")
  // do something with s
  s
 })    

df.withColumn("md5",md5(array($"x",$"y",$"z")))

甚至传递整行，如果您有混合类型的列，这也可以：

val md5 = udf((r:Row) => {
  val s = r.mkString(",")
  // do something with s
  s
 })

df.withColumn("md5",md5(struct($"x",$"y",$"z")))

【讨论】：

【解决方案3】：

如果您想使用自定义分隔符连接所有列，请使用：

df.withColumn('row_hash', md5(concat_ws('||', *df.columns)))

对于计算行哈希很有用。

【讨论】：