spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Adrian Tanase <atan...@adobe.com>
Subject Re: Job splling to disk and memory in Spark Streaming
Date Wed, 21 Oct 2015 09:19:58 GMT
+1 – you can definitely make it work by making sure you are using the same partitioner (including
the same number of partitions).

For most operations like reduceByKey, updateStateByKey – simply specifying it enough.

There are some gotchas for other operations:

  *   mapValues and flatMapValues preserve partitioning
  *   map and flatMap don’t as they can’t be sure of your logic. If you are absolutely
sure that the emitted values will remain on the same partition then you can also override
the partitioner to avoid shuffle
  *   Union on 2 Dstreams throws away partitioning. Again, if you know that it’s safe to
do it, then you need to look at transformWith and push down to RDD.union which preserves partitioning

By using these tricks I’ve successfully forced a pretty complex streaming pipeline (including
2 updateStateByKey, unions, flatmaps, repartitions, custom partitioner, etc) to execute in
a single stage.

Hope this helps,
-adrian

From: Tathagata Das
Date: Wednesday, October 21, 2015 at 10:36 AM
To: swetha
Cc: user
Subject: Re: Job splling to disk and memory in Spark Streaming

Well, reduceByKey needs to shutffle if your intermediate data is not already partitioned in
the same way as reduceByKey's partitioning.

reduceByKey() has other signatures that take in a partitioner, or simply number of partitions.
So you can set the same partitioner as your previous stage. Without any further insight into
the structure of your code its hard to say anything more.

On Tue, Oct 20, 2015 at 5:59 PM, swetha <swethakasireddy@gmail.com<mailto:swethakasireddy@gmail.com>>
wrote:
Hi,

Currently I have a job that has spills to disk and memory due to usage of
reduceByKey and a lot of intermediate data in reduceByKey that gets
shuffled.

How to use custom partitioner in Spark Streaming for  an intermediate stage
so that  the next stage that uses reduceByKey does not have to do shuffles?

Thanks,
Swetha



--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Job-splling-to-disk-and-memory-in-Spark-Streaming-tp25149.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.

---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org<mailto:user-unsubscribe@spark.apache.org>
For additional commands, e-mail: user-help@spark.apache.org<mailto:user-help@spark.apache.org>


Mime
View raw message