Filtering dataframe array items based on an external array with intersection

Question

I'm trying to define a way to filter elements from WrappedArrays in DFs. The filter is based on an external list of elements.

Looking for a solutions I found this question. It is very similar, but it seems not to work for me. I'm using Spark 2.4.0. This is my code:

val df = sc.parallelize(Array((1, Seq("s", "v", "r")),(2, Seq("r", "a", "v")),(3, Seq("s", "r", "t")))).toDF("foo","bar")


def filterItems(flist: Seq[String]) = udf {
  (recs: Seq[String]) => recs match {
    case null => Seq.empty[String]
    case recs => recs.intersect(flist)
  }}

df.withColumn("filtercol", filterItems(Seq("s", "v"))(col("bar"))).show(5)

My expected result would be:

+---+---------+---------+ 
|foo| bar|filtercol| 
+---+---------+---------+ 
| 1 |[s, v, r]|   [s, v]| 
| 2 |[r, a, v]|      [v]| 
| 3| [s, r, t]|      [s]| 
+---+---------+---------+

But I'm getting this error:

java.lang.ClassCastException: cannot assign instance of scala.collection.immutable.List$SerializationProxy to field org.apache.spark.rdd.RDD.org$apache$spark$rdd$RDD$$dependencies_ of type scala.collection.Seq in instance of org.apache.spark.rdd.MapPartitionsRDD

abiratsis · Accepted Answer · 2019-05-14 21:08:34Z

3

You can use the build-in function in Spark 2.4 without too much effort actually:

import org.apache.spark.sql.functions.{array_intersect, array, lit}

val df = sc.parallelize(Array((1, Seq("s", "v", "r")),(2, Seq("r", "a", "v")),(3, Seq("s", "r", "t")))).toDF("foo","bar")

val ar = Seq("s", "v").map(lit(_))
df.withColumn("filtercol", array_intersect($"bar", array(ar:_*))).show

Output:

+---+---------+---------+
|foo|      bar|filtercol|
+---+---------+---------+
|  1|[s, v, r]|   [s, v]|
|  2|[r, a, v]|      [v]|
|  3|[s, r, t]|      [s]|
+---+---------+---------+

The only tricky part is Seq("s", "v").map(lit(_)) which will map each string into lit(i). The intersection function accepts two arrays. The first one is the value of bar column. The second one is created it on the fly with array(ar:_*), which will contain values of lit(i).

edited May 14, 2019 at 21:08

answered May 14, 2019 at 15:44

abiratsis

7,3414 gold badges31 silver badges49 bronze badges

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

Comments

Raphael Roth · Accepted Answer · 2019-05-14 11:39:02Z

0

If you pass an attribute of ArrayType into an UDF, it arrives as an instance of WrappedArray, which is not a List. So you should change recs type to Seq, IndexedSeq or WrappedArray, normally I just use plain Seq:

def filterItems(flist: List[String]) = udf {
  (recs: Seq[String]) => recs match {
    case null => Seq.empty[String]
    case recs => recs.intersect(flist)
  }}

answered May 14, 2019 at 11:39

Raphael Roth

27.3k19 gold badges98 silver badges152 bronze badges

1 Comment

pez_betta Over a year ago

You are right about this. But it doesn't solve my problem. I could change of all List types to Seq types and error would still be the same. I have edited the question so this is not the issue. Thanks for your help.

Collectives™ on Stack Overflow

Filtering dataframe array items based on an external array with intersection

2 Answers 2

Comments

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related