Spark с использованием python: как разрешить Stage x содержит задачу с очень большим размером (xxx KB). Максимальный рекомендуемый размер задачи - 100 КБ

Я только что создал список python range(1,100000).

Использование SparkContext выполнило следующие шаги:

a = sc.parallelize([i for i in range(1, 100000)])
b = sc.parallelize([i for i in range(1, 100000)])

c = a.zip(b)

>>> [(1, 1), (2, 2), -----]

sum  = sc.accumulator(0)

c.foreach(lambda (x, y): life.add((y-x)))

Что дает предупреждение следующим образом:

ARN TaskSetManager: Этап 3 содержит задачу с очень большим размером (4644 КБ). Максимальный рекомендуемый размер задачи - 100 КБ.

Как разрешить это предупреждение? Есть ли способ справиться с размером? А также это повлияет на временную сложность больших данных?

Ответ 1

Развертывание комментария @leo9r: рассмотрим использование не python range, а sc.range https://spark.apache.org/docs/1.6.0/api/python/pyspark.html#pyspark.SparkContext.range.

Таким образом, вы избегаете переноса огромного списка от вашего драйвера к исполнителям.

Конечно, такие RDD обычно используются только для тестирования, поэтому вы не хотите, чтобы их транслировали.

Ответ 2

Искры отправляет копию каждой переменной во время доставки задачи. Для больших размеров таких переменных вы можете использовать Broadcast Variables

Если вы все еще сталкиваетесь с проблемами размера, то, возможно, эти данные должны быть сами по себе RDD

изменить: обновить ссылку

Ответ 3

Общая идея заключается в том, что PySpark создает столько java-процессов, сколько есть исполнителей, а затем отправляет данные в каждый процесс. Если процессов слишком мало, узкое место памяти происходит в области кучи java.

В вашем случае особая ошибка заключается в том, что RDD, созданный с помощью sc.parallelize([...]), не указывал число разделов (аргумент numSlices, см. docs). И RDD по умолчанию имеет несколько разделов, которые слишком малы (возможно, он состоит из одного раздела).

Чтобы решить эту проблему, просто укажите количество желаемых разделов:

a = sc.parallelize([...], numSlices=1000)   # and likewise for b

Как вы указываете большее и большее количество срезов, вы увидите уменьшение размера, указанного в предупреждающем сообщении. Увеличьте количество срезов, пока не получите больше предупреждающего сообщения. Например, получение

Stage 0 contains a task of very large size (696 KB). The maximum recommended task size is 100 KB

означает, что вам нужно указать больше фрагментов.

Еще один совет, который может быть полезен при работе с проблемами памяти (но это не связано с предупреждающим сообщением): по умолчанию память, доступная каждому исполнителю, составляет 1 ГБ или около того. Вы можете указать большие суммы через командную строку, например, с помощью --executor-memory 64G.