scala spark使用spark shell中的udf函数在dataframe列中进行数组操作

Question

scala/火花在火花外壳中使用udf函数在数据框列中进行数组操作

< code>df.printSchema

root
|-- x: timestamp (nullable = true)
|-- date_arr: array (nullable = true)
|    |-- element: timestamp (containsNull = true)

样本数据：

|x                      | date_arr                                                              |  
|---------------------- |---------------------------------------------------------------------- |  
| 2009-10-22 19:00:00.0 | [2009-08-22 19:00:00.0, 2009-09-19 19:00:00.0, 2009-10-24 19:00:00.0] |  
| 2010-10-02 19:00:00.0 | [2010-09-25 19:00:00.0, 2010-10-30 19:00:00.0]                        |  

在udf.jar，我有这个函数来获取上限日期在date_arr根据x：

class CeilToDate extends UDF {
  def evaluate(arr: Seq[Timestamp], x: Timestamp): Timestamp = {
    arr.filter(_.before(x)).last
  }
}

添加jar到火花外壳：火花外壳--jarsudf.jar

在火花外壳，我有HiveContext作为val hc=new HiveContext（spc），并创建函数：hc.sql（"创建临时函数ceil_to_date'com.abc.udf.CeilToDate'"）

当我进行查询时：hc.sql（“选择ceil_to_date（date_arr，x）作为来自df的ceildate”）.显示，期望有一个这样的数据帧：

|ceildate              |        
|----------------------|  
|2009-09-19 19:00:00.0 |  
|2010-09-25 19:00:00.0 |  

但是，它会抛出以下错误：

组织 apache.spark.sql.AnalysisException： 没有用于 Hive udf 类 com.abc.udf.CeilToDate 的处理程序，因为： 没有匹配方法用于类 com.abc.udf.CeilToDate 与 （数组，时间戳）。可能的选项： FUNC（结构

Answer 1

为什么要经历创建udf jar并将jar包含在spack-shell中的所有复杂性。您可以在spack-shell中创建一个并在您的数据框架中使用它。

假设您的数据帧为

scala> df.show(false)
+---------------------+---------------------------------------------------------------------+
|x                    |date_arr                                                             |
+---------------------+---------------------------------------------------------------------+
|2009-10-22 19:00:00.0|[2009-08-22 19:00:00.0, 2009-09-19 19:00:00.0, 2009-10-24 19:00:00.0]|
|2010-10-02 19:00:00.0|[2010-09-25 19:00:00.0, 2010-10-30 19:00:00.0]                       |
+---------------------+---------------------------------------------------------------------+

您可以在spark shell中创建udf函数，但在此之前，您需要三个导入。

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

scala> import java.sql.Timestamp
import java.sql.Timestamp

scala> import scala.collection._
import scala.collection._

然后你可以创建一个udf函数

scala> def ceil_to_date = udf((arr: mutable.WrappedArray[Timestamp], x: Timestamp) => arr.filter(_.before(x)).last)
ceil_to_date: org.apache.spark.sql.expressions.UserDefinedFunction

您所需的输出＜code＞数据帧

scala> df.select(ceil_to_date(col("date_arr"), col("x")).as("ceildate")).show(false)
+---------------------+
|ceildate             |
+---------------------+
|2009-09-19 19:00:00.0|
|2010-09-25 19:00:00.0|
+---------------------+

我希望答案是有帮助的