You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by mtimper <mi...@timper.com> on 2014/11/18 01:40:31 UTC
RDD Blocks skewing to just few executors
Hi I'm running a standalone cluster with 8 worker servers.
I'm developing a streaming app that is adding new lines of text to several
different RDDs each batch interval. Each line has a well randomized unique
identifier that I'm trying to use for partitioning, since the data stream
does contain duplicates lines. I'm doing partitioning with this:
val eventsByKey = streamRDD.map { event => (getUID(event), event)}
val partionedEventsRdd = sparkContext.parallelize(eventsByKey.toSeq)
.partitionBy(new HashPartitioner(numPartions)).map(e => e._2)
I'm adding to the existing RDD like with this:
val mergedRDD = currentRDD.zipPartitions(partionedEventsRdd, true) {
(currentIter,batchIter) =>
val uniqEvents = ListBuffer[String]()
val uids = Map[String,Boolean]()
Array(currentIter, batchIter).foreach { iter =>
iter.foreach { event =>
val uid = getUID(event)
if (!uids.contains(uid)) {
uids(uid) = true
uniqEvents +=event
}
}
}
uniqEvents.iterator
}
val count = mergedRDD.count
The reason I'm doing it this way is that when I was doing:
val mergedRDD = currentRDD.union(batchRDD).coalesce(numPartions).distinct
val count = mergedRDD.count
It would start taking a long time and a lot of shuffles.
The zipPartitions approach does perform better, though after running an hour
or so I start seeing this
in the webUI.
<http://apache-spark-user-list.1001560.n3.nabble.com/file/n19112/Executors.png>
As you can see most of the data is skewing to just 2 executors, with 1
getting more than half the Blocks. These become a hotspot and eventually I
start seeing OOM errors. I've tried this a half a dozen times and the 'hot'
executors changes, but not the skewing behavior.
Any idea what is going on here?
Thanks,
Mike
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/RDD-Blocks-skewing-to-just-few-executors-tp19112.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: RDD Blocks skewing to just few executors
Posted by Sean Owen <so...@cloudera.com>.
Hm is data locality a factor here? I don't know.
Just a side note: this doesn't cause OOM errors per se since the cache
won't exceed the % of heap it's allowed. However that will hasten OOM
problems due to tasks using too much memory, of course. The solution
is to get more memory to the tasks or reduce their working set size.
On Fri, Mar 20, 2015 at 12:32 PM, Alessandro Lulli <lu...@di.unipi.it> wrote:
> Hi All,
>
> I'm experiencing the same issue with Spark 120 (not verified with previous).
>
> Could you please help us on this?
>
> Thanks
> Alessandro
>
> On Tue, Nov 18, 2014 at 1:40 AM, mtimper <mi...@timper.com> wrote:
>>
>> Hi I'm running a standalone cluster with 8 worker servers.
>> I'm developing a streaming app that is adding new lines of text to several
>> different RDDs each batch interval. Each line has a well randomized unique
>> identifier that I'm trying to use for partitioning, since the data stream
>> does contain duplicates lines. I'm doing partitioning with this:
>>
>> val eventsByKey = streamRDD.map { event => (getUID(event), event)}
>> val partionedEventsRdd = sparkContext.parallelize(eventsByKey.toSeq)
>> .partitionBy(new HashPartitioner(numPartions)).map(e => e._2)
>>
>> I'm adding to the existing RDD like with this:
>>
>> val mergedRDD = currentRDD.zipPartitions(partionedEventsRdd, true) {
>> (currentIter,batchIter) =>
>> val uniqEvents = ListBuffer[String]()
>> val uids = Map[String,Boolean]()
>> Array(currentIter, batchIter).foreach { iter =>
>> iter.foreach { event =>
>> val uid = getUID(event)
>> if (!uids.contains(uid)) {
>> uids(uid) = true
>> uniqEvents +=event
>> }
>> }
>> }
>> uniqEvents.iterator
>> }
>>
>> val count = mergedRDD.count
>>
>> The reason I'm doing it this way is that when I was doing:
>>
>> val mergedRDD = currentRDD.union(batchRDD).coalesce(numPartions).distinct
>> val count = mergedRDD.count
>>
>> It would start taking a long time and a lot of shuffles.
>>
>> The zipPartitions approach does perform better, though after running an
>> hour
>> or so I start seeing this
>> in the webUI.
>>
>>
>> <http://apache-spark-user-list.1001560.n3.nabble.com/file/n19112/Executors.png>
>>
>> As you can see most of the data is skewing to just 2 executors, with 1
>> getting more than half the Blocks. These become a hotspot and eventually I
>> start seeing OOM errors. I've tried this a half a dozen times and the
>> 'hot'
>> executors changes, but not the skewing behavior.
>>
>> Any idea what is going on here?
>>
>> Thanks,
>>
>> Mike
>>
>>
>>
>>
>> --
>> View this message in context:
>> http://apache-spark-user-list.1001560.n3.nabble.com/RDD-Blocks-skewing-to-just-few-executors-tp19112.html
>> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>>
>> ---------------------------------------------------------------------
>> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
>> For additional commands, e-mail: user-help@spark.apache.org
>>
>
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: RDD Blocks skewing to just few executors
Posted by Alessandro Lulli <lu...@di.unipi.it>.
Hi All,
I'm experiencing the same issue with Spark 120 (not verified with previous).
Could you please help us on this?
Thanks
Alessandro
On Tue, Nov 18, 2014 at 1:40 AM, mtimper <mi...@timper.com> wrote:
> Hi I'm running a standalone cluster with 8 worker servers.
> I'm developing a streaming app that is adding new lines of text to several
> different RDDs each batch interval. Each line has a well randomized unique
> identifier that I'm trying to use for partitioning, since the data stream
> does contain duplicates lines. I'm doing partitioning with this:
>
> val eventsByKey = streamRDD.map { event => (getUID(event), event)}
> val partionedEventsRdd = sparkContext.parallelize(eventsByKey.toSeq)
> .partitionBy(new HashPartitioner(numPartions)).map(e => e._2)
>
> I'm adding to the existing RDD like with this:
>
> val mergedRDD = currentRDD.zipPartitions(partionedEventsRdd, true) {
> (currentIter,batchIter) =>
> val uniqEvents = ListBuffer[String]()
> val uids = Map[String,Boolean]()
> Array(currentIter, batchIter).foreach { iter =>
> iter.foreach { event =>
> val uid = getUID(event)
> if (!uids.contains(uid)) {
> uids(uid) = true
> uniqEvents +=event
> }
> }
> }
> uniqEvents.iterator
> }
>
> val count = mergedRDD.count
>
> The reason I'm doing it this way is that when I was doing:
>
> val mergedRDD = currentRDD.union(batchRDD).coalesce(numPartions).distinct
> val count = mergedRDD.count
>
> It would start taking a long time and a lot of shuffles.
>
> The zipPartitions approach does perform better, though after running an
> hour
> or so I start seeing this
> in the webUI.
>
> <
> http://apache-spark-user-list.1001560.n3.nabble.com/file/n19112/Executors.png
> >
>
> As you can see most of the data is skewing to just 2 executors, with 1
> getting more than half the Blocks. These become a hotspot and eventually I
> start seeing OOM errors. I've tried this a half a dozen times and the 'hot'
> executors changes, but not the skewing behavior.
>
> Any idea what is going on here?
>
> Thanks,
>
> Mike
>
>
>
>
> --
> View this message in context:
> http://apache-spark-user-list.1001560.n3.nabble.com/RDD-Blocks-skewing-to-just-few-executors-tp19112.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
>