Почему не очень большая сцена Spark, использующая всех доступных исполнителей?

Question

Я выполняю задание Spark с очень большими этапами (например,> 20 тыс. Заданий) и выполняю его с исполнителями от 1 до 2 тыс.

В некоторых случаях этап может показаться нестабильным: многие доступные исполнители со временем простаивают, несмотря на то, что все еще находятся в середине этапа со многими незавершенными задачами. С точки зрения пользователя, кажется, что задачи заканчиваются, но исполнители, которые завершили данную задачу, не получают новую задачу, назначенную им. В результате этап занимает больше времени, чем нужно, и много часов ЦП исполнителя тратится на работу на холостом ходу.

Пример журнала Spark stderr в нестабильный период - обратите внимание, что количество запущенных задач со временем уменьшается до тех пор, пока оно почти не достигает нуля, а затем внезапно возвращается к> 1k запущенных задач:

[Stage 0:==============================>                 (17979 + 1070) / 28504]
[Stage 0:==============================>                 (18042 + 1019) / 28504]
[Stage 0:===============================>                 (18140 + 921) / 28504]
[Stage 0:===============================>                 (18222 + 842) / 28504]
[Stage 0:===============================>                 (18263 + 803) / 28504]
[Stage 0:===============================>                 (18282 + 786) / 28504]
[Stage 0:===============================>                 (18320 + 751) / 28504]
[Stage 0:===============================>                 (18566 + 508) / 28504]
[Stage 0:================================>                (18791 + 284) / 28504]
[Stage 0:================================>                (18897 + 176) / 28504]
[Stage 0:================================>                (18940 + 134) / 28504]
[Stage 0:================================>                (18972 + 107) / 28504]
[Stage 0:=================================>                (19035 + 47) / 28504]
[Stage 0:=================================>                (19067 + 17) / 28504]
[Stage 0:================================>               (19075 + 1070) / 28504]
[Stage 0:================================>               (19107 + 1039) / 28504]
[Stage 0:================================>                (19165 + 982) / 28504]
[Stage 0:=================================>               (19212 + 937) / 28504]
[Stage 0:=================================>               (19251 + 899) / 28504]
[Stage 0:=================================>               (19355 + 831) / 28504]
[Stage 0:=================================>               (19481 + 708) / 28504]

Вот как выглядит stderr, когда этап работает стабильно - количество запущенных задач остается примерно постоянным, потому что новые задачи назначаются исполнителям, когда они заканчивают свои предыдущие задачи:

[Stage 1:===================>                            (11599 + 2043) / 28504]
[Stage 1:===================>                            (11620 + 2042) / 28504]
[Stage 1:===================>                            (11656 + 2044) / 28504]
[Stage 1:===================>                            (11692 + 2045) / 28504]
[Stage 1:===================>                            (11714 + 2045) / 28504]
[Stage 1:===================>                            (11741 + 2047) / 28504]
[Stage 1:===================>                            (11771 + 2047) / 28504]
[Stage 1:===================>                            (11818 + 2047) / 28504]

При каких обстоятельствах это произойдет, и как я могу избежать такого поведения?

Примечание: я использую динамическое распределение, но я почти уверен, что это не связано с этой проблемой - например, в течение нестабильного периода, в пользовательском интерфейсе Spark Application Master, я вижу, что ожидаемое число исполнителей «Активно». ", но не запущены" Активные задачи ".

Answer 1

Я видел такое поведение от искры, когда количество времени, затрачиваемое на одно задание, очень мало.По какой-то причине планировщик, по-видимому, предполагает, что задание будет выполняться быстрее без дополнительных накладных расходов на распространение, поскольку каждое задание выполняется так быстро.

Несколько вещей, которые нужно попробовать:

• Попробуйте .coalesce() уменьшить количество разделов, чтобы каждый раздел работал дольше (предоставлено, это может привести к шагу случайного воспроизведения и может увеличить общее время работы, вам придется истечь)
• Tweakspark.locality.wait* настройки здесь .Если каждая задача занимает меньше времени ожидания по умолчанию, равного 3s, то, возможно, планировщик просто пытается заполнить существующие слоты и никогда не имеет возможности выделить больше слотов.

У меня естьеще предстоит отследить точно что является причиной этой проблемы, так что это всего лишь предположения и догадки, основанные на моих собственных наблюдениях в моем собственном (гораздо меньшем) кластере.