从用户定义的功能创建一个Spark DataFrame列

Question

这应该很琐碎 - 但是我缺少一些东西：

import org.apache.spark.sql._
import org.apache.spark.functions._
val rand = new java.util.Random(42)
val zipUdf = udf( () => 10000 +  rand.nextInt(200))
val zdf = df.withColumn("postalCode",col(zipUdf))

直接使用udf显然是不正确的：

found   : org.apache.spark.sql.expressions.UserDefinedFunction
 required: org.apache.spark.sql.Column

需要什么魔术？

Update ：我已经尝试了以下内容：

 val zdf = df.withColumn("postalCode",zipUdf)

真正的问题是zipUdf()不是zipUdf

Answer 1

它可以无障碍工作：

val someDF = Seq((0, "hello"), (1, "world")).toDF("id", "text")    
import org.apache.spark.sql.functions.udf   
val rand = new java.util.Random(42)
val zipUdf = udf(() => 10000 +  rand.nextInt(200))
someDF.withColumn("postalCode", zipUdf()).show

上述代码的输出：

+---+-----+----------+
| id| text|postalCode|
+---+-----+----------+
|  0|hello|     10130|
|  1|world|     10163|
+---+-----+----------+

与withColumn一起使用col()是失败的主要原因。

Answer 2

您无需将其包装在col（）中。以下代码应起作用。

val zdf = df.withColumn("postalCode",zipUdf())

Answer 3

问题（如错误状态）是zipUdf是类型UserDefinedFunction，而withColumn函数期望Column。您需要实际调用zipUdf函数才能返回Column，即使用()：

val zdf = df.withColumn("postalCode", zipUdf())