spark : duplicate stages while using join

Question

I am using the spark java api and I have come to notice this weird thing that I can't explain. As you can see in

which is the dag visualization of my program execution, no other stage uses the computation of stage 3 , also the three operation in stage 3 are exactly the first 3 operations of stage 2, so my question , why is stage 3 computed separately ? I have also run the program without the last join operation , which gives the following dag,

notice here that there is no parallel stage like in the previous one . I believe that due to this unexplained stage 3 , my program is slowing down .

PS : I am very new to spark and this is my first stackoverflow question, please let me know if it's off topic or requires more detail.

Answer 1

It looks like the join operation takes 2 inputs:

1. the result of the map which is the third operation on stage 2
2. the result of the flatMap which is the 6th operation of phase 2

Spark looks at a single stage at a time when it plans it's computations. It doesn't know that it will need to keep both sets of values. This can result in the 3rd step values being overwritten while the later steps are being computed. When it gets to the join which requires both sets of values, it will realize that it needs those missing values and will recompute them, which is why you're seeing the additional stage that reproduces the first part of stage 2.

You can tell it to save the intermediate values for later, by calling .cache() on the resulting RDD from the map operation and then joining on the RDD returned from the cache(). This will cause spark to make a best effort to maintain those values in memory. You may still see the new stage appear, but it should complete instantaneously if there was enough available memory to store the values.