无法将函数应用于 Spark Dataframe 列

Question

我正在尝试将函数应用于我的数据框列之一以转换值。列中的值类似于“20160907”，我需要值为“2016-09-07”。

我写了一个这样的函数：

def convertDate(inDate:String ): String = {
   val year = inDate.substring(0,4)
   val month = inDate.substring(4,6)
   val day = inDate.substring(6,8)

   return year+'-'+month+'-'+day
}

在我的 spark scala 代码中，我使用的是这个：

def final_Val {
  val oneDF = hiveContext.read.orc("/tmp/new_file.txt")
  val convertToDate_udf = udf(convertToDate _)
  val convertedDf = oneDF.withColumn("modifiedDate", convertToDate_udf(col("EXP_DATE")))
  convertedDf.show()
}

令人惊讶的是，在 spark shell 中，我能够毫无错误地运行。在 Scala IDE 中，我收到以下编译错误：

Multiple markers at this line:
not enough arguments for method udf: (implicit evidence$2: 
reflect.runtime.universe.TypeTag[String], implicit evidence$3: reflect.runtime.universe.TypeTag[String])org.apache.spark.sql.UserDefinedFunction. Unspecified value parameters evidence$2, evidence$3.

我正在使用 Spark 1.6.2、Scala 2.10.5

有人可以告诉我我在这里做错了什么吗？

我在这篇文章中尝试过使用不同功能的相同代码：stackoverflow.com/questions/35227568/applying-function-to-spark-dataframe-column”。 这段代码我没有遇到任何编译问题。我无法找出我的代码的问题

Answer 1

根据我在 spark-summit 课程中学到的知识，您必须尽可能使用sql.functions 方法。在实现您自己的udf 之前，您必须检查sql.functions 包中是否没有现有的function 可以执行相同的工作。使用现有的函数 spark 可以为您做很多优化，并且它没有义务从 JVM 对象序列化和反序列化您的数据。

为了达到你想要的结果，我会提出这个解决方案：

val oneDF = spark.sparkContext.parallelize(Seq("19931001", "19931001")).toDF("EXP_DATE")
val convertedDF = oneDF.withColumn("modifiedDate", from_unixtime(unix_timestamp($"EXP_DATE", "yyyyMMdd"), "yyyy-MM-dd"))
convertedDF.show()

这给出了以下结果：

+--------+------------+
|EXP_DATE|modifiedDate|
+--------+------------+
|19931001|  1993-10-01|
|19931001|  1993-10-01|
+--------+------------+

希望对您有所帮助。最好的问候