How do I call a UDF on a Spark DataFrame using JAVA?

by

Spark >= 2.3

Scala-style udf can be invoked directly:

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

Spark < 2.3

Even if we assume that your UDF is useful and cannot be replaced by a simple getItem call it has incorrect signature. Array columns are exposed using Scala WrappedArray not plain Java Arrays so you have to adjust the signature:

UDF1 mode = new UDF1<Seq<String>, String>() {
  public String call(final Seq<String> types) throws Exception {
    return types.headOption();
  }
};

If UDF is already registered:

sqlContext.udf().register("mode", mode, DataTypes.StringType);

you can simply use callUDF (which is a new function introduced in 1.5) to call it by name:

df.select(callUDF("mode", col("vs"))).show();

You can also use it in selectExprs:

df.selectExpr("mode(vs)").show();

More Related Contents:

  1. How to handle this scenario in spark? [closed]
  2. How to do aggregate functions, may columns and extract back
  3. How to flatten a struct in a Spark dataframe?
  4. How to create SparkSession with Hive support (fails with “Hive classes are not found”)?
  5. Spark sql how to explode without losing null values
  6. Spark Strutured Streaming automatically converts timestamp to local time
  7. spark broadcast variable Map giving null value
  8. How to set timezone to UTC in Apache Spark?
  9. Change output filename prefix for DataFrame.write()
  10. Retain keys with null values while writing JSON in spark
  11. How to use Column.isin in Java?
  12. Why does SparkSession execute twice for one action?
  13. Spark functions vs UDF performance?
  14. How to define and use a User-Defined Aggregate Function in Spark SQL?
  15. Read whole text files from a compression in Spark
  16. java.lang.ClassCastException using lambda expressions in spark job on remote server
  17. SparkSQL: How to deal with null values in user defined function?
  18. Defining a UDF that accepts an Array of objects in a Spark DataFrame?
  19. Passing a data frame column and external list to udf under withColumn
  20. About how to add a new column to an existing DataFrame with random values in Scala
  21. What are the Spark transformations that causes a Shuffle?
  22. Spark spark-submit –jars arguments wants comma list, how to declare a directory of jars?
  23. Derive multiple columns from a single column in a Spark DataFrame
  24. Using UDF ignores condition in when
  25. How to get element by Index in Spark RDD (Java)
  26. How to save models from ML Pipeline to S3 or HDFS?
  27. Converting mysql table to spark dataset is very slow compared to same from csv file
  28. Difference between SparkContext, JavaSparkContext, SQLContext, and SparkSession?
  29. Spark: get number of cluster cores programmatically
  30. How to call a user defined Matlab from Java using matlabcontrol.jar

Leave a Comment