SPARK: YARN убивает контейнеры для превышения пределов памяти

Question

В настоящее время мы сталкиваемся с проблемой, когда рабочие места Spark видят, как несколько контейнеров погибают за превышение пределов памяти при работе на YARN.

16/11/18 17:58:52 WARN TaskSetManager: Lost task 53.0 in stage 49.0 (TID 32715, XXXXXXXXXX): 
    ExecutorLostFailure (executor 23 exited caused by one of the running tasks) 
    Reason: Container killed by YARN for exceeding memory limits. 12.4 GB of 12 GB physical memory used. 
    Consider boosting spark.yarn.executor.memoryOverhead. 

Следующие аргументы пропускают через искру подать:

--executor-memory=6G 
--driver-memory=4G 
--conf "spark.yarn.executor.memoryOverhead=6G"` 

Я использую Спарк 2.0.1.

Мы увеличили значение memoryOverhead до этого значения после прочтения нескольких сообщений о контейнерах для уничтожения YARN (например, How to avoid Spark executor from getting lost and yarn container killing it due to memory limit?).

Учитывая мои параметры и сообщение журнала, кажется, что «Yarn убивает исполнителей, когда его использование памяти больше, чем (executor-memory + executor.memoryOverhead)».

Нецелесообразно продолжать увеличивать эти накладные расходы в надежде, что в итоге мы найдем значение, при котором эти ошибки не будут возникать. Мы рассматриваем эту проблему на нескольких разных работах. Я был бы признателен за любые предложения по параметрам, которые я должен изменить, вещи, которые я должен проверить, где я должен начинать искать отладки и т. Д. Могу предоставить дополнительные параметры конфигурации и т. Д.

Answer 1

Вы можете уменьшить использование памяти в следующих конфигурациях в spark-defaults.conf:

spark.default.parallelism 
spark.sql.shuffle.partitions 

И есть разница, когда вы используете более 2000 разделов для spark.sql.shuffle.partitions. Вы можете увидеть его в коде искры на Github:

private[spark] object MapStatus { 

    def apply(loc: BlockManagerId, uncompressedSizes: Array[Long]): MapStatus = { 
    if (uncompressedSizes.length > 2000) { 
     HighlyCompressedMapStatus(loc, uncompressedSizes) 
    } else { 
     new CompressedMapStatus(loc, uncompressedSizes) 
    } 
} 

Я рекомендую попробовать использовать более 2000 Перегородки для испытания. Это может быть быстрее, когда вы используете очень большие наборы данных. А согласно this ваши задачи могут быть короткими, как 200 мс. Правильную конфигурацию найти нелегко, но в зависимости от вашей рабочей нагрузки она может изменить время.