Spark DataFrame UDF Partitioning Columns

Question

I want to transform a column. The new column should only contain a partition of the original column. I defined the following udf:

def extract (index : Integer) = udf((v: Seq[Double]) => v.grouped(16).toSeq(index))

To use it in a loop later with

myDF = myDF.withColumn("measurement_"+i,extract(i)($"vector"))

The original vector column was created with:

var vectors :Seq[Seq[Double]] = myVectors
vectors.toDF("vector")

But in the end I get the following error:

Failed to execute user defined function(anonfun$user$sparkapp$MyClass$$extract$2$1: (array<double>) => array<double>)

Have I defined the udf incorrectly?

akuiper · Accepted Answer · 2017-09-11 02:51:39Z

3

I can reproduce the error when I try to extract the elements that don't exist, i.e. give an index that is larger than the sequence length:

val myDF = Seq(Seq(1.0, 2.0 ,3, 4.0), Seq(4.0,3,2,1)).toDF("vector")
myDF: org.apache.spark.sql.DataFrame = [vector: array<double>]

def extract (index : Integer) = udf((v: Seq[Double]) => v.grouped(2).toSeq(index))
// extract: (index: Integer)org.apache.spark.sql.expressions.UserDefinedFunction

val i = 2

myDF.withColumn("measurement_"+i,extract(i)($"vector")).show

Gives this error:

org.apache.spark.SparkException: Failed to execute user defined function($anonfun$extract$1: (array<double>) => array<double>)

Most likely you have the same problem while doing toSeq(index), try use toSeq.lift(index) which returns None if the index is out of bound:

def extract (index : Integer) = udf((v: Seq[Double]) => v.grouped(2).toSeq.lift(index))
extract: (index: Integer)org.apache.spark.sql.expressions.UserDefinedFunction

Normal index:

val i = 1    
myDF.withColumn("measurement_"+i,extract(i)($"vector")).show
+--------------------+-------------+
|              vector|measurement_1|
+--------------------+-------------+
|[1.0, 2.0, 3.0, 4.0]|   [3.0, 4.0]|
|[4.0, 3.0, 2.0, 1.0]|   [2.0, 1.0]|
+--------------------+-------------+

Index out of bound:

val i = 2
myDF.withColumn("measurement_"+i,extract(i)($"vector")).show
+--------------------+-------------+
|              vector|measurement_2|
+--------------------+-------------+
|[1.0, 2.0, 3.0, 4.0]|         null|
|[4.0, 3.0, 2.0, 1.0]|         null|
+--------------------+-------------+

edited Sep 11, 2017 at 2:51

answered Sep 10, 2017 at 16:54

akuiper

216k33 gold badges362 silver badges379 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

user4054919 Over a year ago

Thanks a lot, the debugging of this error has cost me a lot of time. +1 for your detailed answer!

Collectives™ on Stack Overflow

Spark DataFrame UDF Partitioning Columns

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related