Как я могу назвать UDF на Spark DataFrame с помощью JAVA?

Тот же вопрос, что и здесь, но у меня недостаточно очков, чтобы комментировать.

Согласно последней документации Spark, udf может использоваться двумя различными способами: один с SQL, а другой с DataFrame. Я нашел несколько примеров того, как использовать udf с SQL, но не смог найти ни одного о том, как использовать udf непосредственно в DataFrame.

Решение, предоставленное __callUDF()__ по вопросу, связанному выше, использует __callUDF()__ который является _deprecated_ и будет удален в Spark 2.0 в соответствии с документацией Spark Java API. Там написано:

"так как это избыточно с помощью udf()"

так что это означает, что я должен иметь возможность использовать __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?

Ответ 1

Spark> = 2,3

Scala-стиль 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();