将函数应用于Spark Dataframe列

Question

作为一个来自R的用户，我习惯于在列上轻松执行操作。是否有一种简单的方法可以将我在Scala中编写的此函数转换为R代码？

def round_tenths_place( un_rounded:Double ) : Double = {
    val rounded = BigDecimal(un_rounded).setScale(1, BigDecimal.RoundingMode.HALF_UP).toDouble
    return rounded
}

将其应用于数据框的一列 - 类似于我希望它所做的：

 bid_results.withColumn("bid_price_bucket", round_tenths_place(bid_results("bid_price")) )

我还没有找到任何简单的方法，正在努力想出如何解决这个问题。一定有比将数据框架转换为RDD，然后从行的RDD中选择正确字段并映射函数至所有值更简单的方法，对吧？同时，创建SQL表并使用SparkSQL UDF来完成这个操作也应该更加简洁，是吗？

- Michael Discenza

---

https://dev59.com/_V4b5IYBdhLWcg3wchXs - Jean Logeart

Answer 1

您可以如下定义UDF：

val round_tenths_place_udf = udf(round_tenths_place _)
bid_results.withColumn(
  "bid_price_bucket", round_tenths_place_udf($"bid_price"))

尽管内置的Round表达式使用的逻辑与您的函数完全相同，并且应该足够，更不用说更加高效：

import org.apache.spark.sql.functions.round

bid_results.withColumn("bid_price_bucket", round($"bid_price", 1))

以下还有相关内容:

• Spark中更新数据框列的方法
• 如何对Spark数据框的一列应用函数？

- zero323

---

我该如何将 $"bid_price" 调用参数化？假设我已经将列名存储在一个变量中，例如：val tg_column = "bid_price"，然后使用 $tg_column 这样的语法。 - Nambu14

这种方法在Databricks中会导致NotSerializableException异常。唯一的解决方法似乎是编写一个箭头函数并将其包装到udf()中。如果我引用任何其他函数，都会出现此异常。 - greatvovan