apache-spark - 如何使用变换高阶函数？

问题描述

这是关于transform高阶函数（https://issues.apache.org/jira/browse/SPARK-23908）。

有什么方法可以将它用作标准功能（在 package 中org.apache.spark.sql.functions._）？

我有一个字符串数组，我想对每个字符串应用 URI 规范化。现在我用 UDF 做到了。我刚刚使用 spark 2.4.0 跳过了 UDF。

正如我所看到的，它应该用于selectExprlikedf.selectExpr("transform(i, x -> x + 1)")但它只意味着与一起使用selectExpr吗？

以这种方式使用它是否可以为转换提供自定义功能？有什么方法可以实现它，还是我应该求助于使用好的旧 UDF？

标签： apache-sparkapache-spark-sql

无论如何将它用作位于包 org.apache.spark.sql.functions._ 中的标准函数？

目前它仅适用于 SQL 表达式，但如果你想返回Column你的 use expr：

org.apache.spark.sql.functions._

expr("transform(i, x -> x + 1)"): Column

以这种方式使用它是否可以为转换提供自定义功能？

可以使用 Scala UDF*：

spark.udf.register("f", (x: Int) => x + 1)

Seq((1, Seq(1, 2, 3))).toDF("id", "xs")
  .withColumn("xsinc", expr("transform(xs, x -> f(x))"))
  .show

+---+---------+---------+
| id|       xs|    xsinc|
+---+---------+---------+
|  1|[1, 2, 3]|[2, 3, 4]|
+---+---------+---------+

尽管它似乎并没有比 UDF 提供任何真正的好处Seq。

* 对 Python UDF 的部分支持似乎已经到位（udf 被识别，类型被正确派生，调用被调度），但是从 2.4.0 开始，序列化机制似乎被破坏了（所有记录都传递给 UDF作为None）：

from typing import Optional
from pyspark.sql.functions import expr

sc.version

'2.4.0'

def f(x: Optional[int]) -> Optional[int]:
    return x + 1 if x is not None else None

spark.udf.register('f', f, "integer")

df = (spark
    .createDataFrame([(1, [1, 2, 3])], ("id", "xs"))
    .withColumn("xsinc", expr("transform(xs, x -> f(x))")))

df.printSchema()

root
 |-- id: long (nullable = true)
 |-- xs: array (nullable = true)
 |    |-- element: long (containsNull = true)
 |-- xsinc: array (nullable = true)
 |    |-- element: integer (containsNull = true)

df.show()

+---+---------+-----+
| id|       xs|xsinc|
+---+---------+-----+
|  1|[1, 2, 3]| [,,]|
+---+---------+-----+

当然，这里没有真正的性能提升潜力 - 它调度到BasePythonRunner所以开销应该与 plain 相同udf。

apache-spark - 如何使用变换高阶函数？

问题描述

解决方案

推荐阅读