last day (17 days later) » 

5:55 PM
1
A: Scala Spark: Split collection into several RDD?

Jason LendermanMaybe something like this would work: def singlePassMultiFilter[T]( rdd: RDD[T], f1: T => Boolean, f2: T => Boolean, level: StorageLevel = StorageLevel.MEMORY_ONLY ): (RDD[T], RDD[T], Boolean => Unit) = { val tempRDD = rdd mapPartitions { iter => val abuf1 = Array...

 
It can be more efficient at a very low level (CPU cache utilization and stuff like that although it can be eaten by ArrayBuffer maintenance) but at the higher level it performs exactly the same amount of work as repeated filter on a cached rdd. Still it looks much better than an accepted answer which actually makes situation worse.
 
The approach I suggested only requires iterating through rdd once, while repeatedly filtering on rdd will necessitate iterating through rdd multiple times. Also, if one knows that the filtering criteria are mutually exclusive (probably true for many uses), then greater efficiency can be obtained by a simply replacing the two if statements with a single if-else statement. Such an optimization would not be possible by repeatedly filtering on rdd.
 
One way or another, for N elements and M conditions, it is still O(NM). If RDD wasn't cached in memory there would be a huge practical difference but otherwise it doesn't really matter. Anyway, if still I think it is much better approach than partitioning hence the upvote.
 
Yes, but there is a cost to just doing the iteration. At the level of machine instructions this involves increasing a counter and comparing against 0 (or, even worse, loading a register, and comparing against that register) for each element in the collection that you're iterating over. If you can do all the work in a single iteration over the collection then you save some computation. It's not a huge savings, but it is noticeable, especially if computations involved in the filtering criteria are not expensive.
 
Yes, although using ArrayBuffers is not cheap either. Moreover with non-exclusive conditions you need more memory to cache tempRDD than rdd. Finally it requires additional transformation and it is quite expensive even if there is almost nothing to do. If I have some spare I'll try to perform some tests but I am not very optimistic.
 
5:55 PM
Assuming a decent implementation ArrayBuffers shouldn't be too bad. I believe adding elements has constant amortized cost, but maybe there's a better way of dealing with that. As for your concern about additional memory usage, if you're not going to cache the resulting RDDs then you wouldn't want to use this approach in the first place. Finally, the final transformation, i.e. the flatMaps are just "sewing" together a bunch of Iterators (one for each split of the data), so that cost of that should pretty negligible compared to iterating through the data multiple times.
 
It is additional compared to rdd.cache(); rdd.filter(cond1); rdd.filte(cond2)
Regarding flatMap I completely agree.
I've been thinking about the cost of running tasks not processing itself.
 
Yes, but not additional compared to rdd.cache(); val filtered1 = rdd.filter(cond1).cache(); val filtered2 = rdd.filter(cond2).cache() If one or both of the filtered rdds are not going to be cached then you would use a different approach. But even there you could still benefit by not iterating through the original RDD multiple times.
 
There is no need to cache filtered1 or filtered2 unless you want to branch further.
 
If there's no need to cache filtered and filtered2, then there are probably more efficient ways of doing the desired computation then creating them in the first place.
My assumption was that the filtered RDDs were going to be cached (though I admit that my example usage is a little misleading on this point.)
 
Not necessarily. If you clean lineage without branching or iterations the only reason for caching is fault-tolerance not performance.
I admit it can be just my serious aversion to for-loops :)
Still it is a huge improvement compared to partitioning which is simply wrong on the multiple levels.
 
6:13 PM
But I'm trying to avoid unnecessary for-loops at the level of machine code, so you should like that! :-) If you do rdd.filter(cond1).reduce(r1) and rdd.filter(cond2).reduce(r2) you are iterating through the elements of rdd twice. Its possible to do this computation in such a manner that you only iterate through rdd once.
 
it should be grouping not partitioning.
 
Yeah, gotcha
 
Once again, nice solution. Now it is time to go back to actual coding :)
 
Yeah, that's what weekends are for, eh? Anyway, nice chatting with you. Thanks for the up-vote.
 

  last day (17 days later) »