sql - 在 Azure Databricks 的 Spark-SQL 中创建用户定义(非临时)函数

Question

也许这很愚蠢，我是一名 Microsoft SQL/C# 开发人员，之前从未真正使用过任何其他 IDE/编写的 JAVA/SCALA。我正在将一些 Azure SQL 查询迁移到 Azure Databricks 解决方案。

似乎没有 TSQL DATEDIFF_BIG 函数 ( https://learn.microsoft.com/en-us/sql/t-sql/functions/datediff-transact-sql?view=sql-server-2017 ) 的等价物

您找到的解决方案是 - 编写您自己的 UDF。

我在 SCALA Notebook 中完成了(见下文)——这对于临时功能来说效果很好。 ( https://docs.databricks.com/spark/latest/spark-sql/language-manual/create-function.html )

这是我找到的最有用的示例 https://github.com/johnmuller87/spark-udf。

有很多临时函数示例，但我没有找到适合非 JAVA/SCALA 开发人员的永久函数示例。

我安装了 SBT(适用于 Windows 的最新版本 - https://www.scala-sbt.org/1.x/docs/Installing-sbt-on-Windows.html)我还安装了Intellj

我为 IBAN 示例运行了 SBT BUILT，但是在将 JAR 上传到我的 Clusterd 之后，我无法获得 SQL 函数，并且函数注册工作。

CREATE FUNCTION ValidateIBAN AS 'com.ing.wbaa.spark.udf.ValidateIBAN' USING JAR 'spark_udf_assembly_0_2_0' --without extension

SELECT ValidateIBAN('NL20INGB0001234567')

错误总是“SQL 语句中的错误:AnalysisException:UDF/UDAF/UDTF ‘com.ing.wbaa.spark.udf.ValidateIBAN’ 没有处理程序；第 1 行 pos 7”

//import org.apache.spark.sql.types._                         // include the Spark Types to define our schema
import org.apache.spark.sql.types.LongType
import org.apache.spark.sql.functions.udf
import java.time.temporal.ChronoUnit;

// Define function to calculate local time offset
def getTimestampDifference(interval: java.lang.String, date1: java.sql.Timestamp, date2: java.sql.Timestamp) : java.lang.Long = {

  //https://docs.oracle.com/javase/8/docs/api/java/sql/Timestamp.html
  //https://spark.apache.org/docs/2.4.0/sql-reference.html
  //https://alvinalexander.com/scala/how-to-use-scala-match-expression-like-switch-case-statement

  interval match
  {
    case "NANOSECOND"=> return ChronoUnit.NANOS.between(date1.toInstant(), date2.toInstant());
    case "MICROSECOND"=> return ChronoUnit.MICROS.between(date1.toInstant(), date2.toInstant());
    case "MILLISECOND"=> return ChronoUnit.MILLIS.between(date1.toInstant(), date2.toInstant()); // date2.getTime() - date1.getTime();
    case "SECOND"=> return ChronoUnit.SECONDS.between(date1.toInstant(), date2.toInstant());
    case "MINUTE"=> return ChronoUnit.MINUTES.between(date1.toInstant(), date2.toInstant());
    case "HOUR"=> return ChronoUnit.HOURS.between(date1.toInstant(), date2.toInstant());
    case "DAY"=> return ChronoUnit.DAYS.between(date1.toInstant(), date2.toInstant());
    case "WEEK"=> return ChronoUnit.WEEKS.between(date1.toInstant(), date2.toInstant());
    case "MONTH"=> return ChronoUnit.MONTHS.between(date1.toInstant(), date2.toInstant());
    case "YEAR"=> return ChronoUnit.YEARS.between(date1.toInstant(), date2.toInstant());
  }
}

spark.udf.register("DATETIMEDIFF", udf(getTimestampDifference(_:java.lang.String, _:java.sql.Timestamp,_:java.sql.Timestamp),LongType))

实际上我需要的是 - 如何将 SCALA 笔记本转换为 SQL 函数，以便我可以在永久 SQL View 中使用它Azure Databricks 集群版本 5.4(包括 Apache Spark 2.4.3、Scala 2.11)

• 实现什么类
• What Method to implement (override in c#) - 还有关于HIVE或SPARK的不同文章
• 如何设置 SBT Built 或以任何其他方式在 Java 存档中编译它，以便我可以成功创建和运行 SQL 函数(仅在 SQL 中，不在 pyhton 代码中，也不在 scala 代码中 - 在 SQL Notebook 中)

谢谢你的帮助

Answer 1

您引用的 Databricks 中的 CREATE FUNCTION 语句实际上是一个 Hive 命令，而不是 Spark，它期望 UDF 类是一个 Hive UDF。

这也是您收到“没有 UDF/UDAF/UDTF 处理程序”错误的原因。您链接的示例实现了一个 Spark UDF，而您需要的是实现一个 Hive UDF。

要创建 Hive UDF，您需要实现一个扩展类 org.apache.hadoop.hive.ql.exec.UDF 的类并实现一个名为 evaluate 的函数。在您的情况下，整个类(class)应如下所示:

class GetTimestampDifference extends UDF {

  def evaluate(interval: java.lang.String, date1: java.sql.Timestamp, date2: java.sql.Timestamp) : java.lang.Long = {

  //https://docs.oracle.com/javase/8/docs/api/java/sql/Timestamp.html
  //https://spark.apache.org/docs/2.4.0/sql-reference.html
  //https://alvinalexander.com/scala/how-to-use-scala-match-expression-like-switch-case-statement

  interval match
  {
    case "NANOSECOND"=> return ChronoUnit.NANOS.between(date1.toInstant(), date2.toInstant());
    case "MICROSECOND"=> return ChronoUnit.MICROS.between(date1.toInstant(), date2.toInstant());
    case "MILLISECOND"=> return ChronoUnit.MILLIS.between(date1.toInstant(), date2.toInstant()); // date2.getTime() - date1.getTime();
    case "SECOND"=> return ChronoUnit.SECONDS.between(date1.toInstant(), date2.toInstant());
    case "MINUTE"=> return ChronoUnit.MINUTES.between(date1.toInstant(), date2.toInstant());
    case "HOUR"=> return ChronoUnit.HOURS.between(date1.toInstant(), date2.toInstant());
    case "DAY"=> return ChronoUnit.DAYS.between(date1.toInstant(), date2.toInstant());
    case "WEEK"=> return ChronoUnit.WEEKS.between(date1.toInstant(), date2.toInstant());
    case "MONTH"=> return ChronoUnit.MONTHS.between(date1.toInstant(), date2.toInstant());
    case "YEAR"=> return ChronoUnit.YEARS.between(date1.toInstant(), date2.toInstant());
  }
}

然后您需要将其编译为 JAR 文件，将其复制到数据 block 文件系统的某处并使用与之前相同的命令创建永久函数(假设您保留 IBAN 示例的 namespace ):

CREATE FUNCTION GetTimestampDifference AS 'com.ing.wbaa.spark.udf.GetTimestampDifference' USING JAR '[path to your jar in dbfs]'

SELECT GetTimestampDifference ("MILLISECOND",cast("2019-07-08 16:07:03.246" as timestamp), cast("2019-07-08 16:07:03.248" as timestamp))

假设您仍在修改开始时使用的 IBAN 示例项目，为了创建 jar 文件，您必须将以下包依赖项添加到 build.sbt 文件中:

"org.apache.spark" %% "spark-hive" % "2.4.3"