Check whether value is key of another pair pyspark

user_9671977 0
Viewed 186

I guess the answer to this question might be too obvious.

Yet I would like to know how to get a list of values that are not present as key for any pair in my RDD e.g. 

pairs = [(3,2),(1,3),(1,4)] to [2,4] 

keys = pairs.keys().distinct()

Now i want to filter the values of pairs and give back only those that are not in keys, something like: 

filteredValues = pairs.values().filter(lambda x: x not in keys)

Error I get: 

Exception: It appears that you are attempting to broadcast an RDD or reference an RDD from an action or transformation. RDD transformations and actions can only be invoked by the driver, not inside of other transformations; for example, rdd1.map(lambda x: rdd2.values.count() * x) is invalid because the values transformation and count action cannot be performed inside of the rdd1.map transformation. For more information, see SPARK-5063.

So I see obviously that the problem is that I use keys (RDD2) inside of pairs' (RDD1) filtering function. The question is, how can I overcome that in still get my filtering right?

Constructive help appreciated. Thanks.

rdd pyspark python
1 Answers

After reading this question I thought of a self-join. Not sure if it's the simplest way to do it though:

  val myRdd = spark.sparkContext.parallelize(Seq((1,1),(2,2),(3,4),(4,5),(5,6),(1,6)))

  val myRdd2 = myRdd.map(x => (x._2, x._1))     // swap values and keys
    .leftOuterJoin(myRdd)                       // and left-join with the original rdd
    .filter(x => x._2._2 == None)
    .map(x => x._1)
    .distinct()

LeftJoin returns these tuples (see below), so last three steps (filter, map, distinct) are needed to filter out the result to get just "6"

...
(1,(1,Some(6)))
(6,(5,None))
(6,(1,None))
(5,(4,Some(6)))
...
user_2823070
Related