0

I have a lookup rdd of size 6000, lookup_rdd: RDD[String]

a1 a2 a3 a4 a5 .....

and another rdd, data_rdd: RDD[(String, Iterable[(String, Int)])]: (id,(item,count)) which has unique ids,

(id1,List((a1,2), (a3,4))) (id2,List((a2,1), (a4,2), (a1,1))) (id3,List((a5,1)))

FOREACH element in lookup_rdd I want to check whether each id has that element or not, if it is there I put the count and if it's not I put 0, and store in a file.

What is the efficient way to achieve this. Is hashing possible? eg. output I want is:

id1,2,0,4,0,0 id2,1,1,0,2,0 id3,0,0,0,0,1

I have tried this:

val headers = lookup_rdd.zipWithIndex().persist()  
val indexing = data_rdd.map{line =>
  val id = line._1
  val item_cnt_list = line._2
  val arr = Array.fill[Byte](6000)(0)
  item_cnt_list.map(c=>(headers.lookup(c._1),c._2))
  }
indexing.collect().foreach(println)

I get the exception:

org.apache.spark.SparkException: RDD transformations and actions can only be invoked by the driver, not inside of other transformations

Nandita Dwivedi
  • 83
  • 2
  • 11

1 Answers1

2

The bad news is that you cannot use an RDD within another.

The good news is that for your use case, assuming that the 6000 entries are fairly small, there is an ideal solution: collect the RDD on the driver, broadcast it back to each node of the cluster and use it within the other RDD as you did before.

val sc: SparkContext = ???
val headers = sc.broadcast(lookup_rdd.zipWithIndex.collect().toMap)
val indexing = data_rdd.map { case (_, item_cnt_list ) =>
  item_cnt_list.map { case (k, v) => (headers.value(k), v) }
}
indexing.collect().foreach(println)
stefanobaghino
  • 11,253
  • 4
  • 35
  • 63
  • thanks for the answer. Have a similar type situation but in addition ..have to update the look up table inside map function. And for the next element i have to do lookup on the updated lookup table.i understand we cannot do this with broadcast.can you please suggest how to approach this.Even a link to the resource would help. Thanks in advance. – Mohan Mar 05 '18 at 13:33
  • I believe you have a better change creating a question for your particular case, sharing the relevant code. Difficult to say without it. – stefanobaghino Mar 05 '18 at 13:41
  • have added a separate question: can you please have a look. :https://stackoverflow.com/questions/49125735/loop-through-dataframe-and-update-the-lookup-table-simultaneously-spark-scala – Mohan Mar 06 '18 at 07:46
  • i understand the whole answer. i have a problem the same as this which is in this link: can you please look at that? (https://stackoverflow.com/questions/60522369/why-i-cant-change-the-property-of-nodes-using-map-function-in-spark) – Hamid Roghani Mar 04 '20 at 10:22