Defining a UDF that accepts an Array of objects in a Spark DataFrame?

by

What you’re looking for is Seq[o.a.s.sql.Row]:

import org.apache.spark.sql.Row

val my_size = udf { subjects: Seq[Row] => subjects.size }

Explanation:

  • Current representation of ArrayType is, as you already know, WrappedArray so Array won’t work and it is better to stay on the safe side.
  • According to the official specification, the local (external) type for StructType is Row. Unfortunately it means that access to the individual fields is not type safe.

Notes:

  • To create struct in Spark < 2.3, function passed to udf has to return Product type (Tuple* or case class), not Row. That’s because corresponding udf variants depend on Scala reflection:

    Defines a Scala closure of n arguments as user-defined function (UDF). The data types are automatically inferred based on the Scala closure’s signature.

  • In Spark >= 2.3 it is possible to return Row directly, as long as the schema is provided.

    def udf(f: AnyRef, dataType: DataType): UserDefinedFunction
    Defines a deterministic user-defined function (UDF) using a Scala closure. For this variant, the caller must specify the output data type, and there is no automatic input type coercion.

    See for example How to create a Spark UDF in Java / Kotlin which returns a complex type?.

More Related Contents:

  1. Derive multiple columns from a single column in a Spark DataFrame
  2. How to query JSON data column using Spark DataFrames?
  3. How to define and use a User-Defined Aggregate Function in Spark SQL?
  4. How to define partitioning of DataFrame?
  5. How do I detect if a Spark DataFrame has a column
  6. Caused by: java.lang.NullPointerException at org.apache.spark.sql.Dataset
  7. How can I pass extra parameters to UDFs in Spark SQL?
  8. SparkSQL: How to deal with null values in user defined function?
  9. How to split a dataframe into dataframes with same column values?
  10. Provide schema while reading csv file as a dataframe
  11. Dropping a nested column from Spark DataFrame
  12. Spark Dataframe :How to add a index Column : Aka Distributed Data Index
  13. How to zip two (or more) DataFrame in Spark
  14. About how to add a new column to an existing DataFrame with random values in Scala
  15. Joining Spark dataframes on the key
  16. Spark UDAF with ArrayType as bufferSchema performance issues
  17. Apache Spark how to append new column from list/array to Spark dataframe
  18. Create new Dataframe with empty/null field values
  19. Provide schema while reading csv file as a dataframe in Scala Spark
  20. Renaming column names of a DataFrame in Spark Scala
  21. Replace missing values with mean – Spark Dataframe
  22. What is going wrong with `unionAll` of Spark `DataFrame`?
  23. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  24. How to get keys and values from MapType column in SparkSQL DataFrame
  25. Spark DataFrames when udf functions do not accept large enough input variables
  26. Filter spark DataFrame on string contains
  27. Spark Scala: How to convert Dataframe[vector] to DataFrame[f1:Double, …, fn: Double)]
  28. Filling gaps in timeseries Spark
  29. What are possible reasons for receiving TimeoutException: Futures timed out after [n seconds] when working with Spark [duplicate]
  30. Why does join fail with “java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]”?

Leave a Comment