Spark splitting a DStream into several RDDs

Question

The same question also applies to splitting an RDD into several new RDDs.

A DStream or RDD contains several different case classes and I need to turn them into separate RDDs based on case class type.

I'm aware of

val newRDD = rdd.filter { a => a.getClass.getSimpleName == "CaseClass1" }

or

val newRDD = rdd.filter { 
  a => a match { 
    case _: CC1 => true
    case _ => false
  }
}

But this requires many runs through the original RDD, one per case class type.

There must be a more concise way to do the above matching filter?
Is there a way to split an rdd into several by the element type with one parallel pass?

maasg maasg · Accepted Answer · 2015-02-05T17:21:42

1) A more concise way of filtering for a given type is to use rdd.collect(PartialFunction[T,U])

The equivalent of

val newRDD = rdd.filter { a => a.getClass.getSimpleName == "CaseClass1" }

would be:

val newRDD = rdd.collect{case c:CaseClass1 => c}

It could even be combined with additional filtering and transformation:

val budgetRDD = rdd.collect{case c:CaseClass1 if (c.customer == "important") => c.getBudget}

rdd.collect(p:PartialFunction[T,U]) should not be confused with rdd.collect() which delivers data back to the driver.

2) To split an RDD (or a DStream for that matter), filter is the way to go. One must remember that an RDD is a distributed collection. Filter will let you apply a function to a subset of that distributed collection, in parallel, over the cluster.

A structural creation of 2 or more RDDs from an original RDD would incur a 1-to-many shuffle stage, which will be substantially more expensive.

Spark splitting a DStream into several RDDs

2 Answers