Почему кэширование небольших RDD Spark требует большого объема памяти в Yarn?

Возможно, вы можете попробовать repartition(n) ваш RDD на меньшее количество n < 100 разделов перед кэшированием. RDD размером ~ 30 ГБ, вероятно, поместится в память десяти исполнителей по 10 ГБ. Хороший обзор управления памятью Spark можно найти здесь. Таким образом, только те исполнители, которые содержат кешированные блоки, будут прикреплены к вашему приложению, а остальные могут быть возвращены YARN через динамическое выделение Spark после spark.dynamicAllocation.executorIdleTimeout (по умолчанию 60 секунд).

В: Можно ли сохранить только память, необходимую для кэшированных RDD в каждом исполнителе, и освободить остальные вместо того, чтобы всегда сохранять изначально установленную память spark.executor.memory = 10G?

Когда Spark использует YARN в качестве механизма выполнения, YARN выделяет контейнеры указанного (по приложению) размера - не менее spark.executor.memory+spark.executor.memoryOverhead, но может быть даже больше в случае pyspark - для всех исполнителей. Сколько памяти Spark фактически использует внутри контейнера, становится неактуальным, поскольку ресурсы, выделенные контейнеру, будут считаться недоступными для других приложений YARN.

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

• Если вы читаете файлы Parquet или CSV, убедитесь, что они имеют одинаковый размер. Запуск repartition() вызывает перемешивание, которое, если данные настолько искажены, может вызвать другие проблемы, если у исполнителей недостаточно ресурсов.
• Кэш не поможет освободить память исполнителям, потому что он не перераспределяет данные
• Не могли бы вы увидеть в разделе «Временная шкала событий на этапах», насколько велики зеленые полосы? Обычно это связано с распределением данных, так что это способ увидеть, сколько данных загружено (пропорционально) для каждой задачи и сколько они делают. Поскольку всем задачам назначена одинаковая память, вы можете графически увидеть, потрачены ли ресурсы впустую (в случае, если есть в основном крошечные полосы и несколько больших полос). Образец потраченных впустую ресурсов можно увидеть на изображении ниже.

Есть разные способы создания равномерно распределенных файлов для вашего процесса. Я упоминаю некоторые возможности, но наверняка их больше:

• Использование Hive и предложения DISTRIBUTE BY: вам нужно использовать поле, которое одинаково сбалансировано, чтобы создать столько файлов (и с надлежащим размером), сколько ожидалось.
• Если процесс создания этих файлов является процессом Spark, считывающим из БД, попробуйте создать столько подключений, сколько файлов вам нужно, и используйте правильное поле для заполнения разделов Spark. Это достигается, как объясняется здесь и здесь со свойствами partitionColumn, lowerBound, upperBound и numPartitions
• Повторное разделение может работать, но посмотрите, имеет ли смысл объединение в вашем процессе или в предыдущем, генерирующем файлы, из которых вы читаете