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

udfudfudf__callUDF()___deprecated___udf()__udf

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?

最佳答案：

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();

火花<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();