如何在数据框中的每一列上运行 udf？

Question

我有一个 UDF：

val TrimText = (s: AnyRef) => {
    //does logic returns string
}

还有一个数据框：

var df = spark.read.option("sep", ",").option("header", "true").csv(root_path + "/" + file)

我想对数据框中每一列的每个值执行TrimText。

但是，问题是，我的列数是动态的。我知道我可以通过df.columns 获取列列表。但我不确定这将如何帮助我解决我的问题。我该如何解决这个问题？

TLDR 问题 - 当数据帧的列数未知时，对数据帧中的每一列执行 UDF

---

尝试使用：

df.columns.foldLeft( df )( (accDF, c) =>
  accDF.withColumn(c, TrimText(col(c)))
)

抛出此错误：

error: type mismatch;
 found   : String
 required: org.apache.spark.sql.Column
accDF.withColumn(c, TrimText(col(c)))

TrimText 假设返回一个字符串，并期望输入是列中的一个值。因此，它将标准化整个数据帧每一行中的每个值。

Answer 1

您可以使用foldLeft 遍历列列表以使用您的UDF 将withColumn 迭代应用到DataFrame：

df.columns.foldLeft( df )( (accDF, c) =>
  accDF.withColumn(c, TrimText(col(c)))
)

Answer 2

>> I would like to perform TrimText on every value in every column in the dataframe.
>> I have a dynamic number of columns.

当 sql 函数可用于修剪 UDF 时，可以看到下面的代码适合您吗？

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

spark.udf.register("TrimText", (x:String) =>  ..... )

val df2 = sc.parallelize(List(
  (26, true, 60000.00),
  (32, false, 35000.00)
)).toDF("age", "education", "income")

val cols2 = df2.columns.toSet
df2.createOrReplaceTempView("table1")

val query = "select " + buildcolumnlst(cols2) + " from table1 "
println(query)
val dfresult = spark.sql(query)
dfresult.show()

def buildcolumnlst(myCols: Set[String]) = {
  myCols.map(x => "TrimText(" + x + ")" + " as " + x).mkString(",") 
}

结果，

select trim(age) as age,trim(education) as education,trim(income) as income from table1 
+---+---------+-------+
|age|education| income|
+---+---------+-------+
| 26|     true|60000.0|
| 32|    false|35000.0|
+---+---------+-------+

Answer 3

val a = sc.parallelize(Seq(("1 "," 2"),(" 3","4"))).toDF()
import org.apache.spark.sql.Column
import org.apache.spark.sql.functions._
def TrimText(s: Column): Column = {
//does logic returns string
  trim(s)
}
a.select(a.columns.map(c => TrimText(col(c))):_*).show