Искра: стратегия перераспределения после чтения текстового файла

Я запустил кластер следующим образом:

/usr/lib/spark/bin/spark-submit --class MyClass --master yarn-cluster--num-executors 3 --driver-memory 10g --executor-memory 10g --executor-cores 4 /path/to/jar.jar

Первое, что я делаю, - это прочитать большой текстовый файл и посчитать его:

val file = sc.textFile("/path/to/file.txt.gz")
println(file.count())

При этом я вижу, что только один из моих узлов фактически просматривает файл и выполняет подсчет (потому что я вижу только одну задачу). Это ожидалось? Должен ли я переделать свой RDD впоследствии или когда я использую функции уменьшения карты, будет ли Spark делать это для меня?

Ответы

Ответ 1

Похоже, вы работаете с файлом gzipped.

Цитата из мой ответ здесь:

Я думаю, что вы столкнулись с довольно типичной проблемой с gzipped файлами, поскольку они не могут быть загружены параллельно. Более конкретно, один gzipped файл не может быть загружен параллельно несколькими задачами, поэтому Spark загрузит его с 1 задачей и, таким образом, предоставит вам RDD с 1 разделом.

Вам нужно явно переделать RDD после его загрузки, чтобы на нем можно было запускать больше задач.

Например:

val file = sc.textFile("/path/to/file.txt.gz").repartition(sc.defaultParallelism * 3)
println(file.count())

Что касается комментариев по вашему вопросу, причина установки minPartitions здесь не помогает, потому что gzipped файл не разделяется, поэтому Spark всегда будет использовать 1 задача прочитать файл.

Если вы устанавливаете minPartitions при чтении обычного текстового файла или файла, сжатого с помощью сплиттируемого формата сжатия, такого как bzip2, вы увидите, что Spark фактически развернет это количество задач параллельно (до количества ядер доступный в вашем кластере) для чтения файла.