Apache Spark 1.6.0, callUDF is failing

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$2.apply(FunctionRegistry.scala:65)
    at org.apache.spark.sql.catalyst.analysis.SimpleFunctionRegistry$$anonfun$2.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

根据您的代码，似乎找不到函数测试，因为 Scala 代码正在尝试进行反射并找到一个名为 test 的函数，该函数接受长函数或您在STEP_EXECUTION_ID列上使用的任何类型作为 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();