在SQLContext之外的Java中创建SparkSQL UDF

Question

问题

我想在Java中创建一个用户定义函数，可以将其称为Apache
Spark运算符链中的Java方法。我在查找不需要UDF存在于SQL查询中的Java示例时遇到了麻烦。

版本号

• Java 8
• 斯卡拉2.10.6
• 为Hadoop 2.6.0预先构建的Apache Spark 1.6.0

我尝试过的方法

我可以用Java成功创建UDF。但是，除非在SQL查询中，否则无法使用它：

import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;

sqlContext.udf().register("udfUppercase",
    (String string) -> string.toUpperCase(), DataTypes.StringType);

DataFrame oldDF = // a simple DataFrame with a "name" column
oldDF.registerTempTable("df");
DataFrame newDF = sqlContext.sql("SELECT udfUppercase(name) AS name_upper FROM df");

我被困在哪里

我希望Java中的非SQL方法调用样式的UDF看起来像这样：

import static org.apache.spark.sql.functions.udf;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.UserDefinedFunction;
import org.apache.spark.sql.types.DataTypes;

UserDefinedFunction udfUppercase = udf(
    (String string) -> string.toUpperCase(), DataTypes.StringType);

DataFrame oldDF = // a simple DataFrame with a "name" column
newDF = oldDF.withColumn("name_upper", udfUppercase(oldDF.col("name")));

编译此代码会导致在以“ UserDefinedFunction”开头的行上出现编译器错误，因此显然我在猜测正确的签名时的尝试是不正确的：

error: no suitable method found for udf((String st[...]ase(),DataType)
    UserDefinedFunction udfUppercase = udf((String string) -> string.toUpperCase(), DataTypes.StringType);
method functions.<RT#1>udf(Function0<RT#1>,TypeTags.TypeTag<RT#1>) is not applicable
    (cannot infer type-variable(s) RT#1
    (argument mismatch; Function0 is not a functional interface
    multiple non-overriding abstract methods found in interface Function0))

对于尝试的每个推断的udf（）签名，此错误都会继续详细介绍。

我需要的

我需要修复Java代码，以便可以定义和使用udfUppercase
UDF，而无需将其嵌入SQL查询中。我感觉好像缺少了一些非常简单，基本的语法（可能是语法y），但是可能完全没有基础。

工作解决方案（由下面的零323提供）

没有注册和使用Java UDF作为Java方法的好方法，但是可以使用callUDF（）将在SQLContext中注册的UDF插入到运算符链中。

import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;

sqlContext.udf().register("udfUppercase",
    (String string) -> string.toUpperCase(), DataTypes.StringType);

DataFrame oldDF = // a simple DataFrame with a "name" column
newDF = oldDF.withColumn("name_upper", callUDF("udfUppercase", oldDF.col("name")));

另外，请确保使用callUDF（），而不要使用具有不同方法签名的已弃用的callUdf（）。

Answer 1

火花 > = 2.3

SPARK-22945（
在function对象中添加Java UDF API ）添加了简化的udfAPI，类似于Scala和Python：

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

UserDefinedFunction udfUppercase = udf(
  (String s) -> s.toUpperCase(), DataTypes.StringType
);

df.select(udfUppercase.apply(col("name")));

火花 <2.3

长话短说的functions.udf方法不是为Java互操作性而设计的。所有变体都需要TypeTags，虽然可以手动生成它们（我很确定我已经见过DanielDarabos展示了如何在SO上进行生成），但是您可能想避免这种情况。

如果出于某种原因要避免在Scala中编写UDF，最简单的方法是注册UDF
并按名称进行调用：

sqlContext.udf().register("udfUppercase",
  (String string) -> string.toUpperCase(), DataTypes.StringType);

df.select(callUDF("udfUppercase", col("name")));