CODEWITHSUNDEEP
scalaapache-spark
Irene
Irene

Reputation: 792

transform rdd into pairRDD

This is a newbie question.

Is it possible to transform an RDD like (key,1,2,3,4,5,5,666,789,...) with a dynamic dimension into a pairRDD like (key, (1,2,3,4,5,5,666,789,...))?

I feel like it should be super-easy but I cannot get how to.

The point of doing it is that I would like to sum all the values, but not the key.

Any help is appreciated.

I am using Spark 1.2.0

EDIT enlightened by the answer I explain my use case deeplier. I have N (unknown at compile time) different pairRDD (key, value), that have to be joined and whose values must be summed up. Is there a better way than the one I was thinking?

Upvotes: 3

Views: 839

Answers (1)

abalcerek
abalcerek

Reputation: 1819

First of all if you just wanna sum all integers but first the simplest way would be:

val rdd = sc.parallelize(List(1, 2, 3))
rdd.cache()
val first = rdd.sum()
val result = rdd.count - first

On the other hand if you want to have access to the index of elements you can use rdd zipWithIndex method like this:

  val indexed = rdd.zipWithIndex()
  indexed.cache()
  val result = (indexed.first()._2, indexed.filter(_._1 != 1))

But in your case this feels like overkill.

One more thing i would add, this looks like questionable desine to put key as first element of your rdd. Why not just instead use pairs (key, rdd) in your driver program. Its quite hard to reason about order of elements in rdd and i cant not think about natural situation in witch key is computed as first element of rdd (ofc i dont know your usecase so i can only guess).

EDIT

If you have one rdd of key value pairs and you want to sum them by key then do just:

val result = rdd.reduceByKey(_ + _)

If you have many rdds of key value pairs before counting you can just sum them up

  val list = List(pairRDD0, pairRDD1, pairRDD2)
  //another pairRDD arives in runtime
  val newList = anotherPairRDD0::list
  val pairRDD = newList.reduce(_ union _)
  val resultSoFar = pairRDD.reduceByKey(_ + _)
  //another pairRDD arives in runtime
  val result = resultSoFar.union(anotherPairRDD1).reduceByKey(_ + _)

EDIT

I edited example. As you can see you can add additional rdd when every it comes up in runtime. This is because reduceByKey returns rdd of the same type so you can iterate this operation (Ofc you will have to consider performence).

Upvotes: 2

Related Questions