3

Consider the following example:

JavaPairRDD<String, Row> R = input.textFile("test").mapToPair(new PairFunction<String, String, Row>() {
        public Tuple2<String, Row> call(String arg0) throws Exception {
            String[] parts = arg0.split(" ");
            Row r = RowFactory.create(parts[0],parts[1]);
            return new Tuple2<String, Row>(r.get(0).toString(), r);
        }}).partitionBy(new HashPartitioner(20));

The code above creates an RDD named R which is partitioned in 20 pieces by hashing on the first column of a txt file named "test".

Consider that the test.txt file is of the following form:

...
valueA1 valueB1
valueA1 valueB2
valueA1 valueB3
valueA1 valueB4
... 

In my context, I have a known value e.g., valueA1 and I want to retrieve all the other values. It is trivial to do it by using the existing filter operation with the specified value. However, I would like to avoid this since essentially the filter operation will be performed on the whole RDD.

Assume that the hash(valueA1)=3, I would like to perform a given operation only on partition 3. More generally, I am interested in dropping/selecting specific partitions from an RDD and perform operations on them.

From the SPARK API it seems that it is not possible directly is there a workaround to achieve the same thing?

1 Answer 1

4

For single keys you can use lookup method:

rdd.lookup("a")

// Seq[Int] = ArrayBuffer(1, 4)

For an efficient lookup you'll need a RDD which is partitioned, for example using HashPartitioner as below.

If you want to simply filter partitions containing specific keys it can be done with mapPartitionsWithIndex:

import org.apache.spark.HashPartitioner

val rdd = sc.parallelize(
  Seq(("a", 1), ("b", 2), ("c", 3), ("a", 4), ("b", 5)
// A particular number is used only to get a reproducible output
)).partitionBy(new HashPartitioner(8))  

val keys = Set("a", "c")
val parts = keys.map(_.## % rdd.partitions.size)

rdd.mapPartitionsWithIndex((i, iter) =>
  if (parts.contains(i)) iter.filter{ case (k, _) => keys.contains(k) }
  else Iterator()
).collect

// Array[(String, Int)] = Array((a,1), (a,4), (c,3))
9
  • Good, answer. I think, I prefer the mapPartitionsWithIndex option since it retains the RDD type and does not necessarily require to bring data to the driver program.
    – zabetak
    Dec 2, 2015 at 17:08
  • 1
    Why did you use .partitionBy(new HashPartitioner(8))? The second param to parallelize could've done it for you (since HashPartitioner is the default partitioner). Dec 9, 2015 at 19:08
  • @JacekLaskowski Only to get a reproducible output.
    – zero323
    Dec 9, 2015 at 19:14
  • 1
    Just use 8 as the second param and you'll also get...reproducible output. Dec 9, 2015 at 19:17
  • @JacekLaskowski On a second thought it won't work without explicit HashPartitioner. As far as I can tell ParallelCollectionRDD doesn't use HashPartioner and computes sliced only based on indices.
    – zero323
    Dec 9, 2015 at 19:36

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.