Let me clarify the performance issue we face, in the current code, if we
set node locality wait to 0 - the scaling is extremely bad due to heavy
data movement at job start (it is worse than non zero node locality value
actually - so it is pointless to set that atleast in our expts).
Particularly when dataset is large. Effectively, almost all schedules end
up being rack local.

Setting it to anything non zero (however small - in our expts even 1ms -
since effectively it is a much higher wait), the cluster utilization drops
drastically.
Before introducing this change, we had experimented with reducing locality
wait to 0ms and 1ms in earlier codebase - and since that did not help
matters, we modified the scheduling ... initially aggression was
effectively any - which improved situation much better, and adding rack
support improved it even further while minimizing io overhead but at
slightly decreased cluster utilization (so it was a tradeoff between both).


Scheduling is something we need to heavily experiment with - considering
the non trivial gains we saw with relatively small changes.





On Sat, Aug 17, 2013 at 1:47 PM, Matei Zaharia <notifications@github.com>wrote:

> In core/src/main/scala/spark/scheduler/cluster/ClusterScheduler.scala:
>
> >          for (i <- 0 until offers.size) {
> > -          val hostPort = offers(i).hostPort
> > -          // DEBUG code
> > -          Utils.checkHostPort(hostPort)
> > -
> > -          val numProcessLocalTasks =  math.max(0, math.min(manager.numPendingTasksForHostPort(hostPort), availableCpus(i)))
> > -          if (numProcessLocalTasks > 0){
> > -            val list = processLocalOffers.getOrElseUpdate(hostPort, new ArrayBuffer[Int])
> > -            for (j <- 0 until numProcessLocalTasks) list += i
> > +          val execId = offers(i).executorId
> > +          val host = offers(i).host
> > +          for (task <- manager.resourceOffer(execId, host, availableCpus(i))) {
> > +            tasks(i) += task
>
> As I said, you can get the same behavior by controlling the different
> timeouts. As far as I understood it, the biggest change for you was
> actually from skipping node locality to go to rack; this can be done easily
> by setting the node locality wait to 0. If you also want to skip process
> locality, you can try that too, though I don't think this is usually a good
> idea. The new code is also simpler than the old aggression code, more
> efficient (doesn't allocate a whole bunch of hashmaps on every resource
> offer), and avoids some real performance regressions I saw with that
> change. If you guys try it out and see some performance problems that can't
> be solved with these settings, I'm happy to revisit it.
>
> BTW, I've also pushed an update for this that now goes through the
> locality levels in order for the whole set of available offers.
>
> —
> Reply to this email directly or view it on GitHub<https://github.com/mesos/spark/pull/828/files#r5828582>
> .
>