如何使用 JAVA 在火花数据帧上调用 UDF？

Question

类似的问题，但没有足够的观点来评论。

根据最新的Spark文档,< code>udf有两种不同的用法，一种用于SQL，另一种用于DataFrame。我找到了许多关于如何在sql中使用< code>udf的例子，但是还没有找到任何关于如何在数据帧中直接使用< code>udf的例子。

o.p.针对上述问题提供的解决方案使用__callUDF（）__，这是_deprecated_，将根据Spark Java API文档在Spark 2.0中删除。在那里，它说：

“因为它与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

火花

Scala样式＜code＞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();

火花

即使我们假设您的UDF很有用并且不能被简单的getItem调用替换，它的签名也不正确。数组列使用ScalaWrapedArray公开，而不是普通的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);

您可以简单地使用call UDF（这是1.5中引入的新函数）按名称调用它：

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

您也可以在选择Exprs中使用它：

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