Вопрос

1. Как Spark определяет количество разделов?
2. Нужно ли где-то явно указывать количество доступных ядер ЦП, чтобы количество разделов было одинаковым (например, аргумент numPartition метода parallelize, но затем нужно обновлять программу при каждом изменении количества ядер)?

Задний план

Установлен кластер Spark как в Окружающая среда без изменений в файлах spark-env.sh, spark-defaults.conf и объекте SparkConf в программах.

Для программы N Queen количество разделов было 2, и только одному узлу были назначены задачи. Для программы подсчета слов количество разделов было 22, и задачи были распределены по всем узлам. Используется spark-submit для обеих программ.

Программы

N Королева

val sparkConf = new SparkConf().setAppName("NQueen").set("spark.files.overwrite", "true")
val sc = new SparkContext(sparkConf)
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
def isSafe(column: Int, placement: List[Int]): Boolean = { ... }
def placeQueensAt(row: Int, placements: Set[List[Int]]): Set[List[Int]] = { ... }

val initial = sc.parallelize(queensAtFirst)
//val initial = sc.parallelize(queensAtFirst, 12)
println("Partitions = %d".format(initial.partitions.size))

val result = initial.flatMap(x => placeQueensAt(1, Set(x))).collect()

Количество слов

val sparkConf = new SparkConf().setAppName("WordCount").set("spark.files.overwrite", "true")
val sc = new SparkContext(sparkConf)
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

val lines = sc.textFile("hdfs:/user/wynadmin/sfpd.csv")
println("Patitions = %d".format(lines.partitions.size))

val words = for (line <- lines; word <- line.split(",") if word.toLowerCase.matches("[a-z]+")) yield (word, 1)
val counts = words.reduceByKey(_ + _)

Окружающая среда

Spark 2.0.1 (3 узла по 4 ЦП каждый) в Ubuntu 14.04.
Автономное развертывание (не YARN и не Mesos)