java - 如何使用 JAVA 在 Spark DataFrame 上调用 UDF？

Question

与here 类似的问题, 但没有足够的分数在那里发表评论。

根据最新的Spark documentation udf 可以以两种不同的方式使用，一种使用 SQL，另一种使用 DataFrame。我发现了多个有关如何将 udf 与 sql 一起使用的示例，但未能找到有关如何直接在 DataFrame 上使用 udf 的任何示例。

o.p.提供的解决方案在上面链接的问题上使用 __callUDF()__ 这是 _deprecated_ 并且根据 Spark Java API 文档将在 Spark 2.0 中删除。在那里，它说:

"since it's redundant with udf()"

所以这意味着我应该能够使用 __udf()__ 来计算我的 udf，但我不知道该怎么做。我没有偶然发现任何阐明 Java-Spark 程序语法的内容。我错过了什么？

import org.apache.spark.sql.api.java.UDF1;
.
.    
UDF1 mode = new UDF1<String[], String>() {
    public String call(final String[] types) throws Exception {
        return types[0];
    }
};

sqlContext.udf().register("mode", mode, DataTypes.StringType);
df.???????? how do I call my udf (mode) on a given column of my DataFrame df?

Answer 1

Spark >= 2.3

Scala风格的udf可以直接调用:

import static org.apache.spark.sql.functions.*;
import org.apache.spark.sql.expressions.UserDefinedFunction;

UserDefinedFunction mode = udf(
  (Seq<String> ss) -> ss.headOption(), DataTypes.StringType
);

df.select(mode.apply(col("vs"))).show();

Spark < 2.3

即使我们假设您的 UDF 是有用的并且不能被简单的 getItem 调用替换，它也有不正确的签名。数组列是使用 Scala WrappedArray 公开的，而不是普通的 Java 数组，因此您必须调整签名:

UDF1 mode = new UDF1<Seq<String>, String>() {
  public String call(final Seq<String> types) throws Exception {
    return types.headOption();
  }
};

如果 UDF 已经注册:

sqlContext.udf().register("mode", mode, DataTypes.StringType);

你可以简单地使用callUDF (这是1.5引入的新功能)按名称调用它:

df.select(callUDF("mode", col("vs"))).show();

你也可以在selectExprs中使用它:

df.selectExpr("mode(vs)").show();