Spark performs poorly when generating non-associate features

Question

I have been using Spark as a tool for my own feature-generation project. For this specific project, I have two data-sources which I load into RDDs as follows:

Datasource1: RDD1 = [(key,(time,quantity,user-id,...)j] => ... => bunch of other attributes such as transaction-id, etc.
Datasource2: RDD2 = [(key,(t1,t2)j)]

In RDD1, time denotes the time-stamp where the event has happened and, in RDD2, denotes the acceptable time-interval for each feature. The feature-key is "key". I have two types of features as follows:

associative features: number of items
non-associative features: Example: unique number of users

For each feature-key, I need to see which events fall in the interval (t1,t2) and then aggregate those things. So, I have a join followed by a reduce operation as follows:

`RDD1.join(RDD2).map((key,(v1,v2))=>(key,featureObj)).reduceByKey(...)`

The initial value for my feature would be featureObj=(0,set([])) where the first argument keeps number of items and the second stores number of unique user ids. I also partition the input data to make sure that RDD1 and RDD2 use the same partitioner.

Now, when I run the job to just calculate the associative feature, it runs very fast on a cluster of 16 m2.xlarge, in only 3 minutes. The minute I add the second one, the computation time jumps to 5min. I tried to add a couple of other non-associate features and, every time, the run-time increases fast. Right now, my job runs in 15minutes for 15 features 10 of them are non-associative. I also tried to use KyroSerializer and persist RDDs in a serialized form but nothing special happened. Since I will be moving to implement more features, this issue seems to become a bottleneck.

PS. I tried to do the same task on a single big host (128GB of Ram and 16 cores). With 145 features, the whole job was done in 10minutes. I am under the impression that the main Spark bottleneck is JOIN. I checked my RDDs and noticed that both are co-partitioned in the same way. As a single job is calling these two RDDs, I presume they are co-located too? However, spark web-console still shows "2.6GB" shuffle-read and "15.6GB" shuffle-write.

Could someone please advise me if I am doing something really crazy here? Am I using Spark for a wrong application? Thanks for the comments in advance.

With best regards,

Ali

So basically you're adding some variant of groupByKey and you see worse performance? It is pretty much expected behavior. — zero323
That's correct; But, my question is something else; I implemented the whole feature-generation pipeline that I discusses above in a single host (128GB of memory with 16 cores) and the whole process (including all non-associative features) was done in 10min. With Spark (16 r3-2xlarge nodes each with 61GB of memory and 8 cores), I get 15 minutes only for 10% of whole features. I know the main bottleneck is in non-associative features. Is this a sign that Spark is not a right tool for this application? I partitioned data to avoid shuffle but still I see shuffling. — gorjida
Well, when you execute your code one a single machine everything is close-to-local so the cost of the shuffle is minimal. Moreover partitioning doesn't remove shuffle it just makes it explicit. Finally co-partitioned != co-located (it's been already covered on SO). There are most likely other problems there as well. — zero323
All your comments make sense. So, I am trying to make a choice between Spark and a load-balancer-type pipeline where I assign partitions of data to each machine where the machine can run local feature generation, without any shuffling. It seems that shuffling is unavoidable when using Spark. I also need to run at least 100 of non-associative features. Given above things, I feel Spark is not a suitable tool for my application? This is something I want to conclude. Does it seem reasonable? — gorjida

Florian Florian · Accepted Answer · 2016-04-03T07:27:33

I noticed poor performance with shuffle operations, too. It turned out that the shuffle ran very fast when data was shuffled from one core to another within the same executor (locality PROCESS_LOCAL), but much slower than expected in all other situations, even NODE_LOCAL was very slow. This can be seen in the Spark UI.

Further investigation with CPU and garbage collection monitoring found that at some point garbage collection made one of the nodes in my cluster unresponsive, and this would block the other nodes shuffling data from or to this node, too.

There are a lot of options that you can tweak in order to improve garbage collection performance. One important thing is to enable early reclamation of humongous objects for the G1 garbage collector, which requires java 8u45 or higher.

In my case the biggest problem was memory allocation in netty. When I turned direct buffer memory off by setting spark.shuffle.io.preferDirectBufs = false, my jobs ran much more stable.

Spark performs poorly when generating non-associate features

1 Answers