Spark DataFrames when udf functions do not accept large enough input variables

by

User defined functions are defined for up to 22 parameters. Only udf helper is define for at most 10 arguments. To handle functions with larger number of parameters you can use org.apache.spark.sql.UDFRegistration.

For example

val dummy = ((
  x0: Int, x1: Int, x2: Int, x3: Int, x4: Int, x5: Int, x6: Int, x7: Int, 
  x8: Int, x9: Int, x10: Int, x11: Int, x12: Int, x13: Int, x14: Int, 
  x15: Int, x16: Int, x17: Int, x18: Int, x19: Int, x20: Int, x21: Int) => 1)

van be registered:

import org.apache.spark.sql.expressions.UserDefinedFunction

val dummyUdf: UserDefinedFunction = spark.udf.register("dummy", dummy)

and use directly

val df = spark.range(1)
val exprs =  (0 to 21).map(_ => lit(1))

df.select(dummyUdf(exprs: _*))

or by name via callUdf

import org.apache.spark.sql.functions.callUDF

df.select(
  callUDF("dummy", exprs:  _*).alias("dummy")
)

or SQL expression:

df.selectExpr(s"""dummy(${Seq.fill(22)(1).mkString(",")})""")

You can also create an UserDefinedFunction object:

import org.apache.spark.sql.expressions.UserDefinedFunction

Seq(1).toDF.select(UserDefinedFunction(dummy, IntegerType, None)(exprs: _*))

In practice having a function with 22 arguments is not very useful and unless you want to use Scala reflection to generate these there are maintenance nightmare.

I would either consider using collections (array, map) or struct as an input or divide this into multiple modules. For example:

val aLongArray = array((0 to 256).map(_ => lit(1)): _*)

val udfWitharray = udf((xs: Seq[Int]) => 1)

Seq(1).toDF.select(udfWitharray(aLongArray).alias("dummy"))

More Related Contents:

  1. How to query JSON data column using Spark DataFrames?
  2. How to define partitioning of DataFrame?
  3. How do I detect if a Spark DataFrame has a column
  4. Caused by: java.lang.NullPointerException at org.apache.spark.sql.Dataset
  5. How to split a dataframe into dataframes with same column values?
  6. Provide schema while reading csv file as a dataframe
  7. Dropping a nested column from Spark DataFrame
  8. Spark Dataframe :How to add a index Column : Aka Distributed Data Index
  9. Defining a UDF that accepts an Array of objects in a Spark DataFrame?
  10. How to zip two (or more) DataFrame in Spark
  11. MatchError while accessing vector column in Spark 2.0
  12. Joining Spark dataframes on the key
  13. Apache Spark how to append new column from list/array to Spark dataframe
  14. Create new Dataframe with empty/null field values
  15. Provide schema while reading csv file as a dataframe in Scala Spark
  16. Renaming column names of a DataFrame in Spark Scala
  17. Derive multiple columns from a single column in a Spark DataFrame
  18. Replace missing values with mean – Spark Dataframe
  19. What is going wrong with `unionAll` of Spark `DataFrame`?
  20. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  21. How to get keys and values from MapType column in SparkSQL DataFrame
  22. Calculate Cosine Similarity Spark Dataframe
  23. Filter spark DataFrame on string contains
  24. Automatically and Elegantly flatten DataFrame in Spark SQL
  25. Spark extracting values from a Row
  26. Append a column to Data Frame in Apache Spark 1.3
  27. Spark / Scala: forward fill with last observation
  28. Apache Spark: Get number of records per partition
  29. Spark Dataframe Random UUID changes after every transformation/action
  30. Spark ML VectorAssembler returns strange output

Leave a Comment