0

First of all, have I fundamentally misunderstood Spark Standalone mode? The official documentation says

The standalone cluster mode currently only supports a simple FIFO scheduler across applications. However, to allow multiple concurrent users, you can control the maximum number of resources each application will use.

I thought that this implied multiple users could have applications running in parallel, submitting jobs to the same Spark Standalone cluster. However, now I am wondering if this was meant to mean that restricting resources would allow multiple users to each run separate Spark Standalone clusters without starving all other users (or just run other programs on the cluster without Spark starving them of resources). Is this the case?

I have Spark set up in Standalone mode on three VMs running Ubuntu. They can all see each other across a NAT network. One of the machines (192.168.56.101) is the master, while the others are slaves (192.168.56.102 and 192.168.56.103).

The Spark version is 2.1.7.

I have a Java app which creates JavaRDD objects in several threads, each calling .collect() in its own thread. I would have thought that this counts as the kind of "job" which can run in parallel for a single Spark Context object (according to https://spark.apache.org/docs/1.2.0/job-scheduling.html).

Each thread gets a JavaRDD object from a synchronized method of a class co-ordinating access to the (single) JavaSparkContext object. The JavaSparkContext is set up without much tweaking. Essentially it is

public synchronized JavaRDD<String> getRdd(List<String> fooList) {
  if (this.javaSparkContext == null) {
    SparkConf sparkConf = new SparkConf();
    sparkConf.set("spark.executor.memory", "500m");
    // There might be a few more settings here such as host name and port, but nothing directly to do with an executor pool or anything, as far as I remember. I don't have the code in front of me while not at work.
    this.javaSparkContext = JavaSparkContext.fromSparkContext(new SparkContext(sparkConf));
  }

  if (this.jobPool == "fooPool") {
    this.jobPool = "barPool";
  } else {
    this.jobPool = "fooPool";
  }

  this.javaSparkContext.setLocalProperty("spark.scheduler.pool", this.jobPool);

  this.javaSparkContext.requestExecutors(1);
  return this.javaSparkContext.parallelize(fooList);
}

The Spark Context object has set up two job pools (as I set it up to), as far as I can tell from the console log:

... INFO scheduler.FairSchedulableBuilder: Created pool fooPool, schedulingMode: FAIR, minShare: 1, weight: 1
... INFO scheduler.FairSchedulableBuilder: Created pool barPool, schedulingMode: FAIR, minShare: 1, weight: 1
... INFO scheduler.FairSchedulableBuilder: Created pool default, schedulingMode: FIFO, minShare: 1, weight: 1

I started many threads, each submitting one .collect() job, alternating between the two FAIR pools. As far as I can tell, these are being allocated to the two pools:

... INFO: scheduler.TaskSchedulerImpl: Adding task set 0.0 with 1 tasks
... INFO scheduler.FairSchedulableBuilder: Added task set TaskSet_0.0 tasks to pool fooPool

and so on, alternating between the two pools.

(The .collect() call is something like

List<String> consoleOutput = getRdd(fooList).cache().pipe("python ./dummy.py").collect();

but again I don't have the code in front of me. It certainly works in the sense that an Executor correctly executes the command.)

However, the client.StandaloneAppClient$ClientEndpoint only ever creates one Executor, which then proceeds to execute all the tasks in barPool then all the tasks in fooPool in serial (but not FIFO). The Worker node VM has 1 core though I set SPARK_EXECUTOR_INSTANCES, SPARK_EXECUTOR_CORES, SPARK_WORKER_INSTANCES, and SPARK_WORKER_CORES to 4, hoping that that would help somehow.

The Master node also has SPARK_EXECUTOR_INSTANCES, SPARK_EXECUTOR_CORES, SPARK_WORKER_INSTANCES, and SPARK_WORKER_CORES set to 4.

It is only ever one of the Worker nodes which responds, and only ever sends one Executor. Both Worker nodes can communicate with the Master - I can turn off one, and the other will take up the next set of jobs which I submit.

The jobs are trivial jobs, each of which delivers a Python script which performs "sleep for some seconds, printing some stuff", and each job takes a single-element RDD, as a proof of concept for a good business reason, as essentially multiple unrelated RDDs would need to be processed in parallel by unrelated Python scripts.

Is there some setting which I have missed? I know that I am misusing Spark in that I am specifically preventing it from parallelizing according to an RDD, but this is set in stone. I am baffled though that only one Worker responds, given that there are many task sets lined up, in multiple job pools. I even call .requestExecutors(1) with every submission, with the console showing

... INFO cluster.StandaloneSchedulerBackend: Requesting 1 additional executor(s) from the cluster manager

but this seems to be totally ignored.

Any advice will be greatly appreciated!

Edit: added Spark version and Java code for method setting up context. Removed idiotic English mistakes introduced by someone who thought that they would "correct" my question by making it grammatically wrong, which were approved by some people who obviously did not read the edit.

3
  • What is your spark submit look like? Could you add that to the question. Jun 16, 2017 at 18:35
  • I have added some snippets of Java code to the question. I have also reverted the incorrect "corrections" of my English grammar made by @ShankarKoirala earlier. Jun 17, 2017 at 7:35
  • Is there any way of preventing rep-harvesting bots from auto-editting questions with stupid "corrections"? That's twice now that the same incorrect changes have been applied to this question. Jun 19, 2017 at 9:34

1 Answer 1

0

As far as I can tell from a lot of research on the Internet and experimenting with my own code, the answer is "Spark does not work that way".

Specifically: 1) There can only be 1 Spark Context per Java Virtual Machine. 2) Per Spark Context, tasks are only ever executed sequentially.

The way which is used by popular Spark cluster managers such as Mesos or Mist, is to prepare several Spark Contexts, each in its own JVM, and tasks are divided among these Spark Contexts.

I could manage to engage a second worker by using a second JVM (in my case, it was by running the same code simultaneously in the Eclipse debugger and in the IntelliJ debugger), but this is just a confirmation of the kind of set-up described above.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.