Планировщик воздушного потока из-за проблем с памятью

Question

Мы экспериментируем с Apache Airflow (версия 1.10rc2, с python 2.7) и разворачиваем его на kubernetes, веб-сервере и планировщике на разные модули, а база данных также использует cloud sql, но мы столкнулись с проблемами памяти с модулем планировщика

На момент ООМ мы работали только с 4 примерами дагов (примерно 20 задач). Память для стручка составляет 1Gib. Я видел в других сообщениях, что задача может потреблять приблизительно 50 МБ памяти при запуске, и все операции задачи находятся в памяти, ничего не записывается на диск, так что это даст уже 1 ГБ.

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

Есть ли какая-либо настройка, кроме уменьшения параллелизма, которая могла бы быть сделана, чтобы уменьшить использование памяти в самом планировщике?

Я не думаю, что наш вариант использования потребовал бы Dask или Celery для горизонтального масштабирования воздушного потока с большим количеством машин для рабочих.

Еще несколько подробностей о конфигурации:

executor = Localexecutor
parallelism = 10
dag_concurrency = 5
max_active_runs_per_dag = 2
workers = 1
worker_concurrency = 16
min_file_process_interval = 1
min_file_parsing_loop_time = 5
dag_dir_list_interval = 30

Дагами, запущенными в то время, были example_bash_operator, example_branch_operator, example_python_operator и один быстрый Dag, который мы разработали.

Все они просто с простыми задачами / операторами, такими как DummyOperators, BranchOperatos, BashOperators, в некоторых случаях, но выполняющими только echo или sleep, и PythonOperators, которые также выполняют только sleep. Всего было бы приблизительно 40 задач, но не все они выполнялись параллельно, потому что некоторые из них были нисходящими, зависимостями и т. Д., И наш параллелизм установлен на 10 с одним рабочим, как описано выше, и dag_concurrency установлен на 5.

Я не вижу ничего ненормального в журналах воздушного потока, и ни в журналах задач.

Запуск только одного из этих пакетов, кажется, что поток воздуха работает соответственно.

Я вижу много процессов планировщика в модуле планировщика, каждый из которых использует 0,2% памяти или больше:

PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
461384 airflow 20 0 836700 127212 23908 S 36.5 0.4 0:01.19 /usr/bin/python /usr/bin/airflow scheduler 461397 airflow 20 0 356168 86320 5044 R 14.0 0.3 0:00.42 /usr/bin/python /usr/bin/airflow scheduler 44 airflow 20 0 335920 71700 10600 S 28.9 0.2 403:32.05 /usr/bin/python /usr/bin/airflow scheduler 56 airflow 20 0 330548 59164 3524 S 0.0 0.2 0:00.02

И это одна из задач, выполняемых с использованием 0,3% памяти:

PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND 462042 airflow 20 0 282632 91120 10544 S 1.7 0.3 0:02.66 /usr/bin/python /usr/bin/airflow run example_bash_operator runme_1 2018-08-29T07:39:48.193735+00:00 --local -sd /usr/lib/python2.7/site-packages/apache_airflow-1.10.0-py2.7.egg/airflow/example_dags/example_bash_operator.py

Answer 1

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

Как вы видели, планировщик создаст несколько процессов ветвления. Также каждая задача (кроме Dummy) будет выполняться в своем собственном процессе. В зависимости от оператора и данных, которые он обрабатывает, объем памяти, необходимый для выполнения задачи, может сильно различаться.

Параметр параллелизма будет напрямую ограничивать количество одновременно выполняемых задач во всех запусках / задачах dag, что будет иметь наиболее драматический эффект для вас при использовании LocalExecutor. Вы также можете попробовать установить max_threads в [scheduler] на 1.

Итак, (очень) общее эмпирическое правило милостиво к ресурсам:

[256 for scheduler itself] + ( [parallelism] * (100MB + [size of data you'll process]) )

Где размер данных необходимо будет изменить в зависимости от того, загружаете ли вы полный набор данных или обрабатываете его фрагменты в ходе выполнения задачи.

Даже если вы не думаете, что вам нужно масштабировать кластер, я все равно рекомендовал бы использовать CeleryExecutor, хотя бы для того, чтобы изолировать планировщик и задачи друг от друга. Таким образом, если ваш планировщик или работник из сельдерея умрет, это не сломит обоих. Особенно работает в k8, если ваш планировщик работает, он убьет его вместе с любыми запущенными задачами. Если вы запускаете их в разных модулях и модуль планировщика перезапускается, вы можете выполнять задачи, которые вы можете выполнять непрерывно. Если у вас будет больше работников, это уменьшит влияние всплесков памяти / обработки на другие задачи.