将两列传递给 scala 中的 udf？

Question

我有一个包含两列的数据框，一列是数据，另一列是该数据字段中的字符数。

Data    Count
Hello   5
How     3
World   5

我想根据计数列中的值更改列数据的值。如何实现？我用 udf 试过了：

invalidrecords.withColumn("value",appendDelimiterError(invalidrecords("value"),invalidrecords("a_cnt")))

这似乎失败了，这是正确的做法吗？

Answer 1

这是一个简单的方法

首先你创建一个 dataframe

import sqlContext.implicits._
val invalidrecords = Seq(
  ("Hello", 5),
  ("How", 3),
  ("World", 5)
).toDF("Data", "Count")

你应该

+-----+-----+
|Data |Count|
+-----+-----+
|Hello|5    |
|How  |3    |
|World|5    |
+-----+-----+

那么你定义udf函数为

import org.apache.spark.sql.functions._
def appendDelimiterError = udf((data: String, count: Int) => "value with error" )

然后您使用 withColumn 作为

调用

invalidrecords.withColumn("value",appendDelimiterError(invalidrecords("Data"),invalidrecords("Count"))).show(false)

你的输出应该是

+-----+-----+----------------+
|Data |Count|value           |
+-----+-----+----------------+
|Hello|5    |value with error|
|How  |3    |value with error|
|World|5    |value with error|
+-----+-----+----------------+

您可以编写逻辑而不是从 udf 函数返回字符串

已编辑

在下面的评论中回答您的要求需要您更改 udf 函数和 withColumn，如下所示

def appendDelimiterError = udf((data: String, count: Int) => {
  if(count < 5) s"convert value to ${data} - error"
  else data
} )

invalidrecords.withColumn("Data",appendDelimiterError(invalidrecords("Data"),invalidrecords("Count"))).show(false)

你的输出应该是

+----------------------------+-----+
|Data                        |Count|
+----------------------------+-----+
|Hello                       |5    |
|convert value to How - error|3    |
|World                       |5    |
+----------------------------+-----+

将两列传递给 scala 中的 udf？

Passing two columns to a udf in scala?

scala

user-defined-functions

apache-spark