My spark app is too slow, how can I increase the speed significantly?
<p>This is part of my spark code which is very slow. By slow I mean for 70 Million data rows it takes almost 7 minutes to run the code but I need it to run in under 5 seconds if possible. I have a cluster with 5 spark nodes with 80 cores and 177 GB memory of which 33Gb are currently used.</p>

<pre><code>range_expr = col("created_at").between(
                            datetime.now()-timedelta(hours=timespan),
                            datetime.now()-timedelta(hours=time_delta(timespan))
                        )
article_ids = sqlContext.read.format("org.apache.spark.sql.cassandra").options(table="table", keyspace=source).load().where(range_expr).select('article','created_at').repartition(64*2)

axes = sqlContext.read.format("org.apache.spark.sql.cassandra").options(table="table", keyspace=source).load()

#article_ids.join(axes,article_ids.article==axes.article)
speed_df = article_ids.join(axes,article_ids.article==axes.article).select(axes.article,axes.at,axes.comments,axes.likes,axes.reads,axes.shares) \
     .map(lambda x:(x.article,[x])).reduceByKey(lambda x,y:x+y) \
     .map(lambda x:(x[0],sorted(x[1],key=lambda y:y.at,reverse = False))) \
     .filter(lambda x:len(x[1])&gt;=2) \
     .map(lambda x:x[1][-1]) \
     .map(lambda x:(x.article,(x,(x.comments if x.comments else 0)+(x.likes if x.likes else 0)+(x.reads if x.reads else 0)+(x.shares if x.shares else 0))))
</code></pre>

<p>I believe especially this part of the code is particularly slow:</p>

<pre><code>  sqlContext.read.format("org.apache.spark.sql.cassandra").options(table="table", keyspace=source).load() 
</code></pre>

<p>When put in spark it transforms into this which I think causes it to be slow :</p>

<pre><code>javaToPython at NativeMethodAccessorImpl.java:-2
</code></pre>

<p><a href="https://i.stack.imgur.com/S6Noa.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/S6Noa.png" alt="This is mu spark ui"></a></p>

<p>Any help would really be appreciated. Thanks</p>

<p><strong>EDIT</strong></p>

<p>The biggest speed problem seems to be JavatoPython. The attached picture is only for part of my data and is already very slow.</p>

<p><a href="https://i.stack.imgur.com/iTJvW.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/iTJvW.png" alt="enter image description here"></a></p>

<p><strong>EDIT (2)</strong></p>

<p>About <code>len(x1)&gt;=2</code>:</p>

<p>Sorry for the long elaboration but I really hope I can solve this problem, so making people understand a quite complex problem in detail is crucial:</p>

<p>this is my rdd example:</p>

<p>rdd1 = [(1,3),(1,5),(1,6),(1,9),(2,10),(2,76),(3,8),(4,87),(4,96),(4,109),(5,10),(6,19),(6,18),(6,65),(6,43),(6,81),(7,12),(7,96),(7,452),(8,59)]</p>

<p>After the spark transformation rdd1 has this form:
rdd_result = [(1,9),(2,76),(4,109),(6,81),(7,452)] 
the result does not contain (3,8),(5,10) because the key 3 or 5 only occur once, I don't want the 3 or 5 to appear.</p>

<p>below is my program:</p>

<p>first:rdd1 reduceByKey then the result is:</p>

<p>rdd_reduceByKey=[(1,[3,5,6,9]),(2,[10,76]),(3,[8]),(4,[87,96,109]),(5,[10]),(6,[19,18,65,43,81]),(7,[12,96,452,59]))]</p>

<p>second:rdd_reduceByKey filter by len(x<a href="https://i.stack.imgur.com/S6Noa.png" rel="nofollow noreferrer">1</a>)>=2 then result is:</p>

<p>rdd_filter=[(1,[3,5,6,9]),(2,[10,76]),(4,[87,96,109]),(6,[19,18,65,43,81]),(7,[12,96,452,59]))]</p>

<p>so the len(x<a href="https://i.stack.imgur.com/S6Noa.png" rel="nofollow noreferrer">1</a>)>=2 is necessary but slow.</p>

<p>Any recommendation improvements would be hugely appreciated.</p>