如何使用scala将特定函数转换为apache spark中的udf函数？ [复制]

Posted 2023-04-14

【中文标题】如何使用scala将特定函数转换为apache spark中的udf函数？ [复制]

【英文标题】：How to convert a specific function to a udf function in apache spark with scala? [duplicate]

【发布时间】：2018-12-13 18:45:26

【问题描述】：

我在 apache spark 中有一个使用 Scala 创建的数据框。此数据框有两列类型为 Array[String]。我编写了一个简单的函数，它采用这两列并返回单词的交集（返回常用单词的数量：Int）。 我的数据框的一个示例如下所示。

data frame example with its columns

功能如下：

def findNumberCommonWordsTitle(string1:Array[String], string2:Array[String]) =
val intersection = string1.intersect(string2)
intersection.length  

我想将此函数转换为 udf 函数。我试过这个：

val fncwt=udf(findNumberCommonWordsTitle(_:Array[String],_:Array[String]))

finalDF.select(fncwt(finalDF("title_from_words"),finalDF("title_to_words"))).show(5)

但我收到如下错误：

error

错误信息说：Caused by: java.lang.ClassCastException: scala.collection.mutable.WrappedArray$ofRef cannot be cast to [Ljava.lang.String;

我做错了什么？我认为问题是类型不匹配，但我不确定。

之后，我想用上面函数的返回值在我的数据框上创建一个新的列。

我怎样才能做到这一点？我应该怎么做才能修复错误？

提前致谢

【参考方案1】：

函数应该是

def findNumberCommonWordsTitle(string1: Seq[String], string2: Seq[String]) =
   ...

参考：https://spark.apache.org/docs/2.3.0/sql-programming-guide.html#data-types

【讨论】：

我已对我的代码进行了此修改，但仍然出现错误！这次有些不同：org.apache.spark.SparkException: Failed to execute user defined function($anonfun$4: (array) => array) 但我现在的函数是：def findNumberCommonWordsTitle(title_from:Seq [String], title_to:Seq[String]) = val intersection = title_from.intersect(title_to) intersection.length

最后，我意识到当我用整数属性调用 .show() 时调用 dataframe.show() 引起的问题，例如2、程序执行。我不明白为什么会这样。

每次我用一个 > 2 的数字调用 .show() 程序都会失败！