Replace some elements of an RDD

Question

I have this file which contains three types of data say A,B,C. I want to replace only Type A elements of this RDD(call it RDD1) with elements of other RDD(call it RDD2) based on some condition.RDD1 and RDD2 have some strings in common.

File structure [RDD[String]]

1 A 2   
2 B 12 13 4
2 C 67 29  
2 A 5

RDD2 [RDD[Row]]

1 A 2 5 6
2 A 5 7 8

I am filtering the first RDD and appending some string to it to create RDD2.

I know that RDDs are immutable but since map function don't take arguments.I am wondering if there is way to achieve this.

EDIT: (considering the comments below)

Sample output RDD[Any]

1 A 2 5 6
2 B 12 13 4
2 C 67 29
2 A 5 7 8

@MukrramRahman '1 A 2 5 6' '2 B 12 13 4' '2 C 67 29 ' '2 A 5 7 8' — Vinay Kumar
And what is the type of your RDDs, I mean is it RDD[Row] or RDD[String] or something else? — Mukrram Rahman
RDD1 is of type RDD[string] and RDD2 is of type RDD[row] and i want to export the output RDD into text file.I think this can be done once we have the final RDD (any type) . — Vinay Kumar
Right now the only solution i can think of is performing an union operation on RDD2 and RDD3(filtering out B and C types ). — Vinay Kumar

Antoni Antoni · Accepted Answer · 2016-02-11T09:04:26

You should consider to use a more appropriate and explicit data structure inside your RDDs, like for example RDD of (key, value) pairs.

Then you can take advantage of the key to perform joins between your RDD1 and RDD2 "a la SQL". I believe this is what Gabber is already doing above, but using the full power of the Scala syntactic sugar.

In a more explicit way: Your initial RDD, as in Gabber:

val rdd1 = sc.parallelize(Seq(List("1", "A", "2"), List("2", "B", "12", "13", "4"), List("2", "C", "67", "29"), List("2", "A", "5")))
val rdd2 = sc.parallelize(Seq(List("1", "A", "2", "5", "6"), List("2", "A", "5", "7", "8")))

Then create with a map a RDD of (key,value) pairs, where the key will be use to satisfy your matching criteria (your key seems to be in your example the first two elements ex. (1,A))

val rdd1KeyValue = rdd1.map(row => ((row(0),row(1)), row)
val rdd2KeyValue = rdd2.map(row => ((row(0),row(1)), row))

Now, since you want perform the "join" for values with the Key "A" and left other non matching, this is a SQL left outer join. So:

val resultRaw = rdd1KeyValue.leftOuterJoin(rdd2KeyValue)

but resultRaw is now, something like:

((2,C),(List(2, C, 67, 29),None))
((1,A),(List(1, A, 2),Some(List(1, A, 2, 5, 6))))
((2,B),(List(2, B, 12, 13, 4),None))
((2,A),(List(2, A, 5),Some(List(2, A, 5, 7, 8))))

So, to take the final result you need to map again to "pick" what you need (the ._1 operator is to take the first value of a (key,value) pair):

val resFinal = result.map(row => row._2._2.getOrElse(row._2._1))

In my case, the final result is:

List(1, A, 2, 5, 6)
List(2, B, 12, 13, 4)
List(2, A, 5, 7, 8)
List(2, C, 67, 29)

Replace some elements of an RDD

2 Answers