JF
я убирада ООМ когда изменяла сеттинги фракшина и фракшин мемори
Size: a a a
2020 October 30
ИГ
На количество Stage в Job
JF
у меня не было ООМ, о которой пишу
JF
у каждой джобы по 2 стейджа
JF
union есть
AS
просто не хочу выкладывать в общий доступ
так замажьте пути к файлам
JF
вот такой даг
GT
Привет всем, пришла за советом/помощью: дано Hadoop Yarn кластер с 50 нодами r5.16xlarge (64 Gb vCPU, ESB storage 500 GB), настройки для экзекьютеров memory 55Gb, по дефолту executor.memort.overhead 10% то есть 5.5 и того 60,5 GB. Получается по 8 экзекьюетеров на одну ноду с памятью 484GB.
Падаем с OutOfMemory Java Heap на операции шафлинга в строяке val totalSize = df.rdd.countApprox(timeout = 200L, confidence = 0.70).getFinalValue().mean, которая нужна чтобы посчитать количество партиций, на которые нужно разбить датафрейм на определенное количество рядков. Операция получается быстрее по сравнению с count. Но к нам может прийти огромные датафреймы, тогда происходит падение и потеря экзекьютеров, а значит дополнительное время к работе джобы.
Когда читаем датафрейм, то персистем его: unified.persist(StorageLevels.MEMORY_AND_DISK). Замена на DISK_ONLY и MEMORY_AND_DISK_SER приводит к замедлению джобы. Мы можем себе позволить, чтобы джоба ехала только 1 час, с такими параметрами она едет около 3 часов и убивается по таймауту. Использую: "spark.io.compression.codec", "zstd", также пробовала изменить дефолтовые на "spark.memory.fraction", "0.4" и "spark.memory.storageFraction", "0.4", чтобы уменьшить cache в памяти и спилить все на диск - на тесте все проходит, ООМ нет, на проде получилось, что отвалилась нода, потому что я забила диск на дозволенный предел + долго едет джоба - что является недопустимым в процессе. На кластере одновременно может ранится таких 4 джобы. Они равномерно +/- разделяют ресурсы кластера. Сейчас все это бежит с ошибками около +/- час. Пока всех устраивает, но я хочу пофиксить с надеждой, что такая джоба будет ранится вместо часа хотя бы 40-45 минут.
Вопросы: 1) стоит ли максимально задать "spark.executor.memory": "58G", тогда executor.memory.overhead будет 5,8 GB и вместе грубо будет 64 GB?
2) взять мощнее машины с большей памятью например r5.24xlarge?
3) val totalSize = df.rdd.countApprox(timeout = 200L, confidence = 0.70).getFinalValue().mean - как-то можно заменить/оптимизировать?
4) в такой ситуации что лучше использовать StorageLevels.MEMORY_AND_DISK или StorageLevels.DISK_ONLY?
5) что еще можно предпринять/попробовать в изменении настроек спарка, чтобы убрать ООМ и при этом не замедлять джобу/ы?
Спасибо заранее за помощь и предложения.
Падаем с OutOfMemory Java Heap на операции шафлинга в строяке val totalSize = df.rdd.countApprox(timeout = 200L, confidence = 0.70).getFinalValue().mean, которая нужна чтобы посчитать количество партиций, на которые нужно разбить датафрейм на определенное количество рядков. Операция получается быстрее по сравнению с count. Но к нам может прийти огромные датафреймы, тогда происходит падение и потеря экзекьютеров, а значит дополнительное время к работе джобы.
Когда читаем датафрейм, то персистем его: unified.persist(StorageLevels.MEMORY_AND_DISK). Замена на DISK_ONLY и MEMORY_AND_DISK_SER приводит к замедлению джобы. Мы можем себе позволить, чтобы джоба ехала только 1 час, с такими параметрами она едет около 3 часов и убивается по таймауту. Использую: "spark.io.compression.codec", "zstd", также пробовала изменить дефолтовые на "spark.memory.fraction", "0.4" и "spark.memory.storageFraction", "0.4", чтобы уменьшить cache в памяти и спилить все на диск - на тесте все проходит, ООМ нет, на проде получилось, что отвалилась нода, потому что я забила диск на дозволенный предел + долго едет джоба - что является недопустимым в процессе. На кластере одновременно может ранится таких 4 джобы. Они равномерно +/- разделяют ресурсы кластера. Сейчас все это бежит с ошибками около +/- час. Пока всех устраивает, но я хочу пофиксить с надеждой, что такая джоба будет ранится вместо часа хотя бы 40-45 минут.
Вопросы: 1) стоит ли максимально задать "spark.executor.memory": "58G", тогда executor.memory.overhead будет 5,8 GB и вместе грубо будет 64 GB?
2) взять мощнее машины с большей памятью например r5.24xlarge?
3) val totalSize = df.rdd.countApprox(timeout = 200L, confidence = 0.70).getFinalValue().mean - как-то можно заменить/оптимизировать?
4) в такой ситуации что лучше использовать StorageLevels.MEMORY_AND_DISK или StorageLevels.DISK_ONLY?
5) что еще можно предпринять/попробовать в изменении настроек спарка, чтобы убрать ООМ и при этом не замедлять джобу/ы?
Спасибо заранее за помощь и предложения.
А чего вы добиваетесь с вычислением количества партиций? Для чего разбиваете именно на это количество
JF
А чего вы добиваетесь с вычислением количества партиций? Для чего разбиваете именно на это количество
чтобы писать csv-и по 10000
JF
некоторые клиенты так просят
JF
я хотела убрать эту операцию, но увы нет
JF
+ равномерно делим
А
можно поставить настройку maxRecordsPerFile, чтобы писать нужное число строк в файл
GT
чтобы писать csv-и по 10000
Может пронумеровать зипвивиндекс (допустим, поле index) и разделить на партиции по полю part = (floor (index/10000))🤔
JF
неуверенна, что это будет быстрее
GT
Быстрее, может быть
GT
неуверенна, что это будет быстрее
Я думаю, избавиться от одного действия и возращения данных на драйвер может помочь, но да, как обычно, зависит от местных реалий данных
JF
спасибо всем :) буду пробовать
AS
тут посередине вызов кеширования?