在 Spark 中调用 Scala UDF 时，如何将 BinaryType 转换为 Array[Byte]？答案

【问题标题】：How can I convert BinaryType to Array[Byte] when calling Scala UDF in Spark?在 Spark 中调用 Scala UDF 时，如何将 BinaryType 转换为 Array[Byte]？
【发布时间】：2021-07-24 13:46:28
【问题描述】：

我在 Scala 中编写了以下 UDF：

import java.io.{ByteArrayOutputStream, ByteArrayInputStream}
import java.util.zip.{GZIPInputStream}

def Decompress(compressed: Array[Byte]): String = {
  val inputStream = new GZIPInputStream(new ByteArrayInputStream(compressed))
  val output = scala.io.Source.fromInputStream(inputStream).mkString
  
  return output
}

val decompressUdf = (compressed: Array[Byte]) => {
  Decompress(compressed)
}

spark.udf.register("Decompress", decompressUdf)

然后我尝试使用以下命令调用 UDF：

val sessionsRawDF =
  sessionRawDF
    .withColumn("WebsiteSession", decompressUdf(sessionRawDF("body")))
    .select(
      current_timestamp().alias("ingesttime"),
      current_timestamp().cast("date").alias("p_ingestdate"),
      col("partition"),
      col("enqueuedTime"),
      col("WebsiteSession").alias("Json")
    )

当我运行它时，我收到以下错误：

command-130062350733681:9：错误：类型不匹配；
找到：org.apache.spark.sql.Column 必需：数组[字节] decompressUdf(col("WebsiteSession")).alias("Json")

在这种情况下，我的印象是 Spark 会隐式获取值并从 spark 类型转到 Array[Byte]。

请帮助我了解发生了什么，我已经为此奋斗了一段时间，不知道还能尝试什么。

【问题讨论】：

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

【解决方案1】：

您需要先将 Scala 函数转换为 Spark UDF，然后才能将其注册为 UDF。例如，

val decompressUdf = udf(Decompress _)

spark.udf.register("Decompress", decompressUdf)

实际上，如果只是在 DataFrame API 中使用 UDF，则无需注册它。您可以简单地运行第一行并使用decompressUdf。仅当您想在 SQL 中使用 UDF 时才需要注册。

【讨论】：