Subtract an RDD from another RDD doesn't work correctly

Question

I want to subtract an RDD from another RDD. I looked into the documentation and I found that subtract can do that. Actually, when I tested subtract, the final RDD remains the same and the values are not removed!

Is there any other function to do that? Or am I using subtract incorrectly?

Here is the code that I used:

 val vertexRDD: org.apache.spark.rdd.RDD[(VertexId, Array[Int])]
 val clusters  = vertexRDD.takeSample(false, 3)
 val clustersRDD: RDD[(VertexId, Array[Int])] = sc.parallelize(clusters)
 val final = vertexRDD.subtract(clustersRDD)
 final.collect().foreach(println(_))

Answer 1

Performing set operations like subtract with mutable types (Array in this example) is usually unsupported, or at least not recommended.

Try using a immutable type instead.

I believe WrappedArray is the relevant container for storing arrays in sets, but i'm not sure.

Answer 2

Recently I tried the subtract operation of 2 RDDs (of array List) and it is working. The important note is - the RDD val after .subtract method should be the list from where you're subtracting, not the other way around.

Correct: val result = theElementYouWantToSubtract.subtract(fromList)

Incorrrect: val reuslt = fromList.subtract(theElementYouWantToSubtract) (will not give any compile/runtime error message)

Answer 3

If your rdd is composed of mutables object it wont work... problem is it wont show an error either so this kind of problems are hard to identify, i had a similar one yesterday and i used a workaround.

rdd.keyBy( someImmutableValue ) -> do this using the same key value to
 both your rdds

val resultRDD = rdd.subtractByKey(otherRDD).values