创建具有非原始数据类型的 UDF 函数并在 Spark-sql 查询中使用：Scala

Question

我正在 scala 中创建一个函数，我想在我的 spark-sql 查询中使用它。我的查询在 hive 中运行良好，或者如果我在 spark sql 中给出相同的查询，但我在多个使用相同的查询地方，所以我想将它创建为可重用的函数/方法，所以只要它需要我就可以调用它。 我在我的 scala 类中创建了以下函数。

def date_part(date_column:Column) = {
    val m1: Column = month(to_date(from_unixtime(unix_timestamp(date_column, "dd-MM-yyyy")))) //give  value as 01,02...etc

    m1 match {
        case 01 => concat(concat(year(to_date(from_unixtime(unix_timestamp(date_column, "dd-MM- yyyy"))))-1,'-'),substr(year(to_date(from_unixtime(unix_timestamp(date_column, "dd-MM-yyyy")))),3,4))
        //etc..
        case _ => "some other logic"
    }
}

但它显示多个错误。

1. 对于 01：

◾十进制整数文字可能没有前导零。 （八进制语法 已经过时了。）

◾类型不匹配；找到：需要 Int(0)：org.apache.spark.sql.Column。

1. 对于“-”：

类型不匹配；发现：需要 Char('-')： org.apache.spark.sql.Column.

1. 对于“substr”：

未找到：值 substr.

另外，如果我创建的任何简单函数也使用类型作为列，我无法注册它，因为我在列格式中无法获得错误。对于所有原始数据类型（String，Long，Int ）它工作正常。但在我的情况下，类型是列所以 我无法做到这一点。有人可以指导我该怎么做。到目前为止，我在 stack-overflow 上发现我需要将此函数与 df 一起使用，然后需要将此 df 转换为临时表。有人可以请指导我任何其他替代方式，以便我现有代码无需太多更改即可使用此功能。

Answer 1

试试下面的代码。

scala> import org.joda.time.format._
import org.joda.time.format._

scala> spark.udf.register("datePart",(date:String) => DateTimeFormat.forPattern("MM-dd-yyyy").parseDateTime(date).toString(DateTimeFormat.forPattern("MMyyyy")))
res102: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,StringType,Some(List(StringType)))

scala> spark.sql("""select datePart("03-01-2019") as datepart""").show
+--------+
|datepart|
+--------+
|  032019|
+--------+

Answer 2

首先，Spark 需要读取一个存储数据的文件，我猜这个文件是 CSV，但你可以使用方法 json insted of csv。

然后您可以添加具有计算值的新列，如下所示：

     import org.apache.spark.sql.functions._

      val df = spark.read
        .option("header", "true")
        .option("inferSchema", "true")
        .csv("/path/mydata.csv")

      def transformDate( dateColumn: String, df: DataFrame) : DataFrame = {
         df.withColumn("calculatedCol", month(to_date(from_unixtime(unix_timestamp(col(dateColumn), "dd-MM-yyyy")))))

         df.withColumn("newColumnWithDate",  when(col("calculatedCol") === "01", concat(concat(year(to_date(from_unixtime(unix_timestamp(col("calculatedCol"), "dd-MM- yyyy"))))-1, lit('-')),substring(year(to_date(from_unixtime(unix_timestamp(col("calculatedCol")), "dd-MM-yyyy"))),4,2))
          .when(col("calculatedCol") === "02","some other logic")
          .otherwise("nothing match")))
      }

     // calling your function for the Dataframe you want transform date column:
     transformDate("date_column", df)

注意有些函数需要一个列作为参数，而不是字符串值，所以使用 lit() 来指定这些值。

不需要UDF（并且在性能方面不推荐），但您可以通过以下方式使用它：

val upper: String => String = _.toUpperCase
import org.apache.spark.sql.functions.udf
val upperUDF = udf(upper)
df.withColumn("upper", upperUDF('text)).show

“上部”函数将是您必须包含转换日期列的逻辑的方法。