Как запустить больше исполнителей в режиме Apache Spark Cluster

Question

У меня 50 рабочих, я хотел бы выполнить свою работу на всех моих рабочих.
В мастер: 8080, я могу видеть все рабочие там,
В мастер: 4040/исполнители, я могу видеть 50 исполнителей,
но когда я запускаю свою работу, информация шоу, как это:

14/10/19 14:57:07 INFO scheduler.TaskSchedulerImpl: Adding task set 0.0 with 1 tasks 
14/10/19 14:57:07 INFO scheduler.TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, slave11, NODE_LOCAL, 1302 bytes) 
14/10/19 14:57:07 INFO nio.ConnectionManager: Accepted connection from [slave11/10.10.10.21:42648] 
14/10/19 14:57:07 INFO nio.SendingConnection: Initiating connection to [slave11/10.10.10.21:54398] 
14/10/19 14:57:07 INFO nio.SendingConnection: Connected to [slave11/10.10.10.21:54398], 1 messages pending 
14/10/19 14:57:07 INFO storage.BlockManagerInfo: Added broadcast_1_piece0 in memory on slave11:54398 (size: 2.4 KB, free: 267.3 MB) 
14/10/19 14:57:08 INFO storage.BlockManagerInfo: Added broadcast_0_piece0 in memory on slave11:54398 (size: 18.4 KB, free: 267.2 MB) 
14/10/19 14:57:12 INFO storage.BlockManagerInfo: Added rdd_2_0 in memory on slave11:54398 (size: 87.4 MB, free: 179.8 MB) 
14/10/19 14:57:12 INFO scheduler.DAGScheduler: Stage 0 (first at GeneralizedLinearAlgorithm.scala:141) finished in 5.473 s 
14/10/19 14:57:12 INFO scheduler.TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 5463 ms on slave11 (1/1) 
14/10/19 14:57:12 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool 

И моя работа подобный код: (командной строки)

master: $ ./spark-shell --master spark://master:7077 

и это (Scala код):

import org.apache.spark.SparkContext 
import org.apache.spark.mllib.classification.SVMWithSGD 
import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics 
import org.apache.spark.mllib.regression.LabeledPoint 
import org.apache.spark.mllib.linalg.Vectors 
import org.apache.spark.mllib.util.MLUtils 

val fileName = "bc.txt" 
val data = sc.textFile(fileName) 

val splits = data.randomSplit(Array(0.9, 0.1), seed = 11L) 
val training = splits(0).cache() 
val test = splits(1) 

val training_1 = training.map { line => 
val parts = line.split(' ') 
LabeledPoint(parts(0).toDouble, Vectors.dense(parts(1).split(' ').map(x => x.toDouble).toArray)) 
} 

val test_1 = test.map { line => 
val parts = line.split(' ') 
LabeledPoint(parts(0).toDouble, Vectors.dense(parts(1).split(' ').map(x => x.toDouble).toArray)) 
} 
val numIterations = 200 

val model = SVMWithSGD.train(training_1, numIterations) 

Мой вопрос, почему только один или два (иногда) запуска задания на моем кластере?
Каким-либо образом можно настроить количество заданий или расписание по планировщику автоматически?
Когда моя работа выполняется по двум задачам, и она будет работать с двумя исполнителями, которые я наблюдаю на главном: 4040,
Это даст 2x ускорение, поэтому я хочу запустить свою работу на всех исполнителей, как я могу это сделать?

Спасибо всем.

Answer 1

Вы можете использовать параметр minPartitions в textFile установить мин число задач, таких как:

val data = sc.textFile(fileName, 10) 

Однако больше разделов, как правило, означает, что больше сетевого трафика, потому что больше разделов сделать Спарк трудно диспетчерских задач к местных исполнителей. Вам необходимо найти остаток minPartitions.