Apache Spark 1.6.0，callUDF 失败

Question

我正在努力使用 callUDF 功能，我总是收到函数未注册的错误。我在下面粘贴了示例代码：

UDF1<String, String> func = new UDF1<String, String>(){
      public String call(String s) throws Exception {
            return s +"fixedString";
  }
};
sqlContext.udf().register("test",func, DataTypes.StringType);
out = out.select(out.col("VERSION"),callUDF("test",out.col("STEP_EXECUTION_ID")) );

我总是收到以下错误，代码中缺少什么。

org.apache.spark.sql.AnalysisException: undefined function test;
    at org.apache.spark.sql.catalyst.analysis.SimpleFunctionRegistry$$anonfun.apply(FunctionRegistry.scala:65)
    at org.apache.spark.sql.catalyst.analysis.SimpleFunctionRegistry$$anonfun.apply(FunctionRegistry.scala:65)
    at scala.Option.getOrElse(Option.scala:120)
    at org.apache.spark.sql.catalyst.analysis.SimpleFunctionRegistry.lookupFunction(FunctionRegistry.scala:64)

Answer 1

根据您的代码，似乎找不到函数 test 因为 Scala 代码正在尝试进行反射并找到一个名为 test 的函数，该函数接受 long 或您在 [=17 上用作 ID 的任何类型=]列。

尝试更改 UDF 的参数类型以匹配列类型。像这样：

public String call(Long id) throws Exception

Answer 2

我已经解决了这个问题，所以如果其他人遇到类似问题，请将其张贴在这里。我有两个问题，1. 在一列中生成 UUID 2. 从列值生成计算值。

问题 1：

import java.util.UUID;

public class RandomGenerator extends scala.runtime.AbstractFunction0<String> {
    public String apply() {
       return UUID.randomUUID().toString();
    }
}

在这种情况下，无需使用 sqlcontext

进行注册

df.withColumn("UUID", callUDF(new RandomGenerator(), DataTypes.StringType)).show();

问题 2：

在这种情况下，可以使用上述方法，或者有人也可以执行以下操作

UDF1< Integer, Integer> func = new UDF1<Integer, Integer>() {
    public Integer call(Integer s) throws Exception {
        return calculate(s);
    }
};

sqlContext.udf().register("calculate", func, DataTypes.IntegerType);
df.select(df.col("calVal"), callUDF("calculate", df.col("value"))).show();

Apache Spark 1.6.0，callUDF 失败

Apache Spark 1.6.0, callUDF is failing

apache-spark

spark-dataframe