{ "version": "https:\/\/jsonfeed.org\/version\/1.1", "title": "Slavlotski: заметки с тегом Airflow", "_rss_description": "Меня зовут Влад и я Data Engineer. В свободное от работы время люблю писать электронную музыку, играть в видеоигры", "_rss_language": "ru", "_itunes_email": "", "_itunes_categories_xml": "", "_itunes_image": "", "_itunes_explicit": "", "home_page_url": "https:\/\/slavlotski.com\/tags\/airflow\/", "feed_url": "https:\/\/slavlotski.com\/tags\/airflow\/json\/", "icon": "https:\/\/slavlotski.com\/pictures\/userpic\/userpic@2x.jpg?1696305806", "authors": [ { "name": "Slavlotski", "url": "https:\/\/slavlotski.com\/", "avatar": "https:\/\/slavlotski.com\/pictures\/userpic\/userpic@2x.jpg?1696305806" } ], "items": [ { "id": "16", "url": "https:\/\/slavlotski.com\/all\/apache-airflow\/", "title": "Apache Airflow", "content_html": "
\n\"\"\n<\/div>\n

Предпоссылки создания Apache Airflow<\/h2>\n

В 2014 году компания Airbnb стремительно развивалась, что привело к увеличению объема данных и усложнению рабочих процессов обработки данных. Управление этими сложными конвейерами данных с помощью традиционных инструментов, таких как Cron и пользовательские скрипты, стало нецелесообразным из-за недостаточной масштабируемости, гибкости и возможностей мониторинга. Airbnb требовалось решение, которое могло бы масштабироваться в соответствии с растущими потребностями в данных.Им требовалось отслеживать состояния рабочих процессов и возможность эффективно отлаживать сбои. Инициатива со стороны Maxime Beauchemin<\/a> привела к созданию масштабируемого, гибкого и удобного инструмента управления рабочими процессами под названием Airflow. Сделав Airflow open-sourceным и наладив взаимодействие с сообществом, Airbnb не только решила свои внутренние проблемы, но и предоставила мощный инструмент широкому сообществу разработчиков. Сегодня Apache Airflow широко используется в различных отраслях индустрии для организации сложных рабочих процессов и конвейеров данных.<\/p>\n

Apache Airflow<\/b> — фреймворк для построения конвейров обработки данных. Airflow сам по себе не является инструментом обработки данных. Он управляет различными компонентами, которые отвечают за обработку ваших данных в конвейерах. Ключевая особенность Airflow заключается в том, что он позволяет легко создавать конвейеры обработки данных, запускаемых по расписанию с помощью языка Python.<\/p>\n

Какие задачи решает Airflow?<\/h2>\n
    \n
  1. Оркестрация рабочих процессов:<\/b> Airflow позволяет управлять зависимостями между задачами, обеспечивая их выполнение в нужном порядке<\/li>\n
  2. Планирование задач:<\/b> С помощью Airflow можно запланировать выполнение задач по заданному расписанию<\/li>\n
  3. Мониторинг и управление задачами: <\/b>Airflow предоставляет интерфейс для отслеживания выполнения задач, просмотра логов и управления задачами в реальном времени<\/li>\n
  4. Интеграция с различными системами:<\/b> Airflow может взаимодействовать с различными источниками данных и системами, такими как базы данных, облачные сервисы, API и другие.<\/li>\n
  5. Автоматизация процессов:<\/b> Airflow помогает автоматизировать повторяющиеся процессы, что снижает ручной труд и минимизирует ошибки.<\/li>\n
  6. Логирование и отслеживание изменений:<\/b> Airflow сохраняет логи выполнения задач, что помогает в отладке и анализе процессов.<\/li>\n<\/ol>\n

    Кому подойдет Airflow<\/h2>\n

    — Для планирования задач, где возможностей Cron стало недостаточно
    \n— У команды уже есть достаточная экспертиза в программировании на Python
    \n— На проекте используется пакетная обработка данных (Batch), а не потоковая (Stream). AirFlow предназначен для Batch-заданий, для потоковой обработки данных лучше использовать Apache NiFi
    \n— Планируется или уже осуществлен переход в облако и необходим надежный оркестратор, поддерживающий все принципы Cloud-Native<\/p>\n

    Конвейеры обработки данных в виде графов<\/h2>\n

    Конвейер обработки данных легко представить в виде графа. В математике граф представляет собой конечный набор узлов с вершинами, соединяющими узлы. В контексте разработки каждый узел в графе представляет собой задачу, а зависимостями между задачами — направленные ребра между узлами. Ребро, направленное от задачи A к задаче B, указывает, что задача A должна быть завершена до того, как может начаться задача B. Такие графы обычно называются ориентированными, или направленными, потому что ребра имеют направление.<\/p>\n

    \n\"\"\n
    Пример ациклического направленного графа. Источник: airflow.apache.org<\/a><\/div>\n<\/div>\n

    Данный тип графа обычно называется ориентированным ациклическим графом, поскольку он содержит ориентированные ребра и у него нет никаких петель или циклов (ациклический).<\/p>\n

    \n\"\"\n
    Пример циклического направленного графа<\/div>\n<\/div>\n

    Анатомия DAG в Airflow<\/h2>\n

    DAG (Directed Acyclic Graph)<\/b> — направленный ациклический граф необходимый для смыслового объединения изолированных задач, которые необходимо выполнить в строго определенной последовательности согласно указанному расписанию. Задачами же в терминологии Airflow называют Task<\/b>. Под задачами понимается, например, загрузка из различных источников, их агрегирование, очистка от дубликатов, сохранение полученных результатов и прочие ETL-процессы. На уровне кода задачи представляют собой Python-функции или Bash-скрипты в терминологии Airflow их называют Operator<\/b>.<\/p>\n

    Task<\/b> — это внутренние компоненты для управления состоянием operator<\/b> и отображения изменений состояния (например, запущено\/завершено) для пользователя. Task управляет выполнением оператора.<\/p>\n

    Operator<\/b> — отвечает за выполнение одной единицы работы. Операторы — это некие готовые шаблоны, которые можно переиспользовать, реализующие логику выполнения (запуск скрипта, команды, обращение к СУБД) с набором параметров на входе. В AirFlow богатый выбор встроенных операторов. Кроме этого, доступно множество специальных операторов — путем установки пакетов поставщиков, поддерживаемых сообществом. Также возможно добавление пользовательских операторов — за счет расширения базового класса BaseOperator.<\/p>\n

    Sensor<\/b> — это особая группа операторов, которые позволяют в случае наступления определенного события запустить выполнение следующих по зависимости задач. В качестве триггера могут выступать получение некоторого файла, готовность данных на источнике и так далее.<\/p>\n

    Так как Airflow оперирует языком Python, то у класса DAG<\/b> могут быть свои экземпляры. В Airflow экземплярами DAGа называют DAG Run<\/b>, связанные с этим DAGом экземляры задач — task Instance<\/b>, а время запуска этих DAGов — execution_date<\/b> (в более поздних версиях Airflow logical_date<\/b>).<\/p>\n

    \n\"\"\n
    Визуализация DAGа. Источник: cloud.vk.com<\/a><\/div>\n<\/div>\n

    Архитектура Airflow и принципы его работы<\/h2>\n

    Web Server <\/b>— компонент, который позволяет визуализировать зависимости в Airflow, анализируемые планировщиком, и предоставляет пользователям основной интерфейс для отслеживания выполнения графов и их результатов. Помимо этого представления, Airflow также предоставляет подробное древовидное представление, в котором показаны все текущие и предыдущие запуски соответствующего DAG. Также здесь приводится беглый обзор того, как работал DAG, и оно позволяет покопаться в задачах, завершившихся сбоем, чтобы увидеть, что пошло не так.<\/p>\n

    \n\"\"\n
    Web UI Airflow. Источник: airflow.apache.org<\/a><\/div>\n<\/div>\n

    Metadata DB (база метаданных)<\/b> — собственный репозиторий метаданных на базе библиотеки SqlAlchemy для хранения глобальных переменных, настроек соединений с источниками данных, статусов выполнения Task Instance, DAG Run и так далее. Хранит состояния scheduler<\/b>, executor<\/b> и веб-сервера<\/b>. Требует установки совместимой с SqlAlchemy базы данных, например, MySQL или PostgreSQL.<\/p>\n

    Scheduler (планировщик)<\/b> — служба, отвечающая за планирование в Airflow и отправку задания на исполнение. Отслеживая все созданные Task и DAG, планировщик инициализирует Task Instance — по мере выполнения необходимых для их запуска условий. По умолчанию раз в минуту планировщик анализирует результаты парсинга директории dags и проверяет, нет ли задач, готовых к запуску.Если все в порядке,то начинает планировать задачи DAG для выполнения, передавая их воркерам<\/b> Airflow. Для выполнения активных задач планировщик использует указанный в настройках Executor<\/b>. Для определенных версий БД (PostgreSQL 9.6+ и MySQL 8+) поддерживается одновременная работа нескольких планировщиков — для максимальной отказоустойчивости.<\/p>\n

    Executor (исполнитель)<\/b>— часть scheduler, механизм, с помощью которого запускаются экземпляры задач. Работает в связке с планировщиком в рамках одного процесса.<\/p>\n

    Worker (рабочий) <\/b>— отдельный процесс, в котором выполняются запланированные задачи. Размещение Worker — локально или на отдельной машине — определяется выбранным типом Executor. Допускается неограниченное число DAG за счет модульной архитектуры и очереди сообщений. Worker могут масштабироваться при использовании Celery или Kubernetes.<\/p>\n

    DAG Directory<\/b> — каталог исполняемых DAGов для остальных компонентов Airflow<\/p>\n

    \n\"\"\n
    Архитектура Apache Airflow. Источник: linkedin.com<\/a><\/div>\n<\/div>\n

    Концепция планирования в Airflow<\/h2>\n

    Выполнение работы с фиксированными интервалами<\/b>
    \nДля многих рабочих процессов, включающих временные процессы, важно знать, в течение какого временного интервала выполняется данная задача. По этой причине Airflow предоставляет задачам дополнительные параметры, которые можно использовать, чтобы определить, в каком интервале выполняется задача. Самый важный из этих параметров — execution_date<\/b>, который обозначает дату и время, в течение которых выполняется DAG<\/b>. execution_date<\/b> (начиная с версии Airflow версии 2.2 logical_date<\/b>) — это не дата, а временная метка, отражающая время начала интервала, для которого выполняется DAG. Время окончания интервала указывается другим параметром, next_execution_date<\/b> (data_interval_end<\/b> в поздних версиях Airflow 2). Вместе эти даты определяют всю продолжительность интервала задачи.<\/p>\n

    \n\"\"\n
    Объяснение интервалов в DAG. Источник: Книга. Apache Airflow и конвейеры обработки данных<\/a><\/div>\n<\/div>\n

    Предположим, что у нас есть DAG, который следует ежедневному интервалу, а затем учитывает соответствующий интервал, в котором должны обрабатываться данные за 2019-01-03. В Airflow этот интервал будет запускаться вскоре после 2019-01-04:00:00, потому что в данный момент мы знаем, что больше не будем получать новые данные за 2019-01-03. Значение переменной execution_date<\/b> при выполнении задач будет 2019-01-03. Это связано с тем, что Airflow определяет дату выполнения DAG как начало соответствующего интервала. Дата выполнения отмечает интервал, а не момент фактического выполнения DAG.<\/p>\n

    Запуск через равные промежутки времени<\/b>
    \nAirflow можно запускать через равные промежутки времени, задав для него запланированный интервал с помощью аргумента schedule_interval<\/b> при инициализации DAG.<\/p>\n

    dag = DAG(\r\ndag_id="dag_1", schedule_interval="@daily" <-- Планируем запуск каждый день в полночь\r\n, start_date=dt.datetime(2019, 1, 1) <-- Дата и время начала планирования запусков DAG\r\n... )<\/code><\/pre>
    Airflow также нужно знать, когда мы хотим начать выполнение DAG, указав дату запуска. Исходя из этой даты, Airflow запланирует первое выполнение нашего DAG. Airflow запускает задачи в конце интервала. Если разработанный DAG вывели на продуктив 1 января 2019 года в 14:00 с start_date — 01-01-2019 и интервалом @daily, то это означает, что первый DAG Run случится в полночь 02-01-2019:00:00.<\/p>\n
    Интервалы на основе Cron<\/b>
    \nДля поддержки более сложных вариантов Airflow позволяет определять интервалы, используя тот же синтаксис, что и у cron<\/b> — планировщика заданий на основе времени, используемого Unix-подобными операционными системами, такими как macOS и Linux. Достаточно указать crontab выражение в аргументе schedule_interval<\/b> при инициализации DAGа.<\/p>\n
    from airflow.models.dag import DAG\r\nimport datetime\r\n\r\ndag = DAG("regular_interval_cron_example", schedule="0 0 * * *", ...) <-- задача будет регулярно запускаться в полночь<\/code><\/pre>
    Частотные интервалы<\/b>
    \nЕсли мы хотим запускать наш DAG раз в три дня, то в качестве интервала можно передать экземпляр timedelta (из модуля datetime в стандартной библиотеке Python) в schedule_interval<\/b>.
    \n— timedelta(minutes=10) каждые 10 минут
    \n— timedelta(hours=2) каждые 2 часа<\/p>\n
    dag = DAG(\r\ndag_id="dag_3", schedule_interval=dt.timedelta(days=3),\r\n......)<\/code><\/pre>
    Правила триггеров<\/h2>\n
    Правила триггеров <\/b>— это, по сути, условия, которые Airflow применяет к задачам, чтобы определить, готовы ли они к выполнению, ориентируясь на их зависимости. Правило триггеров по умолчанию — это all_success<\/b>, которое гласит, что все зависимости задачи должны быть успешно завершены, прежде чем саму задачу можно будет выполнить.<\/p>\n
    А что произойдет, если одна из наших задач обнаружит ошибку во время исполнения? Это означает, что нижестоящую задачу уже нельзя выполнить, поскольку для ее успешного выполнения необходима успешное выполнение предшествующей задачи и она примет статус upstream_failed<\/b>. Более подробно со статусами задач можно ознакомиться в официальной документации Airflow<\/a>.<\/p>\n
    Airflow поддерживает и другие правила триггеров, которые допускают различные типы поведения при ответе на успешные, неудачные или пропущенные задачи. Правило none_failed<\/b> означает, что оно допускает как успешные, так и пропущенные задачи, по-прежнему ожидая завершения всех вышестоящих задач, перед тем как продолжить. Например, правило all_done<\/b> можно использовать для определения задач, которые завершили свое выполнение независимо от их конечного состояния. Более подробно об остальных правилах можно познакомиться в официальной документации Airflow<\/a><\/p>\n
    \n\"\"\n
    Web UI статусов задач. Источник: airflow.apache.org<\/a><\/div>\n<\/div>\n
    Запуск задачи после определенного действия<\/h2>\n
    Действия, связанные с запуском, часто являются результатом внешних событий. Представьте себе файл, который публикуется на общий диск, готовность данных за конкретную дату во внешней БД и т. д. Для решения подобных задач используют сенсоры<\/b>, представляющие собой особый тип (подкласс) операторов. Сенсоры непрерывно опрашивают определенные условия, чтобы определить их истинность, и если условие истинно, то сенсор завершает свою работу со статусом успешно. В противном случае сенсор будет ждать и повторять попытку до тех пор, пока условие не будет истинно, или время ожидания истечет.<\/p>\n
    from airflow.sensors.filesystem import FileSensor\r\n\r\nwait_for_supermarket_1 = FileSensor(\r\ntask_id="wait_for_data", \r\nfilepath="data.csv",\r\n)<\/code><\/pre>
    Здесь FileSensor<\/b> выполняет проверку на предмет наличия файла data.csv и возвращает true, если файл существует. В противном случае возвращается false, и сенсор будет ждать в течение заданного периода (по умолчанию 60 секунд) и повторит попытку.<\/p>\n
    Вывод сенсоров можно посмотреть в логах задач:<\/p>\n
    {file_sensor.py:60} INFO – Poking for file data.csv \r\n{file_sensor.py:60} INFO – Poking for file data.csv \r\n{file_sensor.py:60} INFO – Poking for file data.csv \r\n{file_sensor.py:60} INFO – Poking for file data.csv \r\n{file_sensor.py:60} INFO – Poking for file data.csv<\/code><\/pre>
    Здесь видно, что примерно раз в минуту сенсор осуществляет покинг<\/b> на предмет наличия определенного файла. Покинг<\/b> — так в Airflow называется запуск сенсора и проверка условия.<\/p>\n
    Зачастую сенсоры добавляются в самое начало DAGа, так как мы хотим запустить последующие манипуляции с данными только в случае их готовности на источнике. Таким образом, сенсоры в начале DAG будут непрерывно выполнять опрос на предмет наличия данных и переходить к следующей задаче после выполнения условия.<\/p>\n
    \n\"\"\n
    Web UI работы сенсоров. Источник: livebook.manning.com<\/a><\/div>\n<\/div>\n
    А что произойдет, если однажды один из источников не предоставит свои данные за ожидаемое время?  По умолчанию сенсоры будут падать, как и зависимые от них задачи со статусом upstream_failed<\/b>. Сенсоры принимают аргумент timeout<\/b>, который содержит максимальное количество секунд, в течение которого сенсор может работать. Если в начале очередного покинга количество секунд превысит число, заданное для timeout<\/b>, то это приведет к падению сенсора.<\/p>\n
    \n\"\"\n
    Web UI упавших сенсоров по timeout. Источник: livebook.manning.com<\/a><\/div>\n<\/div>\n
    Количество задач (на разных уровнях), которые Airflow может обработать ограничено. Важно понимать, что в Airflow есть ограничения на максимальное параллельное количество выполняемых задач на разных уровнях; количество задач на каждый DAG, количество задач на глобальном уровне Airflow, количество запусков DAG на каждый DAG и т. д. У класса DAG есть аргумент concurrency<\/b>, определяющий, сколько параллельных задач разрешено в рамках этого DAG. Таким образом каждый сенсор — это одна задача и в случае если у вас их много и значение concurrency<\/b> меньше кол-ва сенсоров, то некоторые сенсоры будут простаивать и не запускать проверку своего условия, эти сенсоры будут ожидать пока освободится слот в DAGе, то есть пока не завершит свое выполнение один из ранее запустившихся сенсоров.<\/p>\n
    Класс сенсора принимает аргумент mode<\/b>, для которого можно задать значение poke<\/b> или reschedule<\/b> (начиная с Airflow версии 1.10.2). По умолчанию задано значение poke<\/b>, что приводит к блокировке. Это означает, что задача сенсора занимает слот, пока выполняется. Время от времени она выполняет покинг, осуществляя проверку условия, а затем ничего не делает, но по-прежнему занимает слот. Режим сенсора reschedule<\/b> освобождает слот после покинга, слот остается занят только когда выполняется проверка.<\/p>\n
    Запуск задач с помощью REST API и интерфейса командной строки<\/h2>\n
    Запуск DAGов можно осуществлять через REST API и командную строку. Это может быть полезно, если вы хотите запускать рабочие процессы за пределами Airflow. Данные, поступающие в случайное время в бакет AWS S3, можно обрабатывать, задав лямбда-функцию для вызова REST API, запуская DAG, вместо того чтобы постоянно запускать опрос с сенсорами.<\/p>\n
    airflow dags trigger dag1<\/code><\/pre>
    Этот код запускает dag1 с датой выполнения, установленной на текущую дату и время. Идентификатор запуска имеет префикс manu­al__<\/b>, указывая на то, что он был запущен вручную или за пределами Airflow.<\/p>\n
    Для обеспечения аналогичного результата можно использовать REST API (например, если у вас нет доступа к командной строки, но к вашему экземпляру Airflow можно подключиться по протоколу HTTP).<\/p>\n
    Преимущества и недостатки Airflow<\/h2>\n
    Преимущества:<\/b>
    \n— Airflow — это фреймворк с открытым исходным кодом
    \n— Конвейеры на основе кода обладают большими возможностями расширения. Преимущество того факта, что Airflow написан на языке Python, состоит в том, что задачи могут выполнять любую операцию, которую можно реализовать на Python. Со временем это привело к разработке множества расширений Airflow, позволяющих выполнять задачи в широком спектре систем, включая внешние базы данных, технологии больших данных и различные облачные сервисы, давая возможность создавать сложные конвейеры обработки данных, объединяющие процессы обработки данных в различных системах
    \n— Конвейеры на основе кода более управляемы: поскольку все находится в коде, он может легко интегрироваться в ваш CI \/ CD управления версиями и общие рабочие процессы разработчика.
    \n— Множество интеграций с разными системами
    \n— Гибкий планировщик
    \n— Возможность масштабирования
    \n— Такие функции, как обратное заполнение (backfill)<\/b>, дают возможность с легкостью (повторно) обрабатывать архивные данные, позволяя повторно вычислять любые производные наборы данных после внесения изменений в код;
    \n— Многофункциональный веб-интерфейс Airflow обеспечивает удобный просмотр результатов работы конвейера и отладки любых сбоев, которые могут произойти
    \n— Поддерживаются все принципы Cloud-Native<\/p>\n
    Недостатки:<\/b>
    \n— Обработка потоковых конвейеров, поскольку Airflow в первую очередь предназначен для выполнения повторяющихся задач по пакетной обработке данных, а не потоковых рабочих нагрузок
    \n— Реализация высокодинамичных конвейеров, в которых задачи добавляются или удаляются между каждым запуском конвейера. Хотя Airflow может реализовать такое динамическое поведение, веб-интерфейс будет показывать только те задачи, которые все еще определены в самой последней версии DAG. Таким образом, Airflow отдает предпочтение конвейерам, структура которых не меняется каждый раз при запуске
    \n— Поддержка Airflow может быстро стать сложной в крупных проектах
    \n— Высокий порог входа в фреймворк<\/p>\n",
            "date_published": "2024-06-30T20:23:16+05:00",
            "date_modified": "2024-09-03T22:38:29+05:00",
            "tags": [
                "Airflow",
                "BigData",
                "IT"
            ],
            "image": "https:\/\/slavlotski.com\/pictures\/apache-airflow-1.png",
            "_date_published_rfc2822": "Sun, 30 Jun 2024 20:23:16 +0500",
            "_rss_guid_is_permalink": "false",
            "_rss_guid": "16",
            "_e2_data": {
                "is_favourite": true,
                "links_required": [
                    "system\/library\/highlight\/highlight.js",
                    "system\/library\/highlight\/highlight.css"
                ],
                "og_images": [
                    "https:\/\/slavlotski.com\/pictures\/apache-airflow-1.png",
                    "https:\/\/slavlotski.com\/pictures\/directed-acyclic-graph.png",
                    "https:\/\/slavlotski.com\/pictures\/directed-cyclic-graph.png",
                    "https:\/\/slavlotski.com\/pictures\/main-definitions-airflow.png",
                    "https:\/\/slavlotski.com\/pictures\/web-server-apache-airflow.jpg",
                    "https:\/\/slavlotski.com\/pictures\/airflow-architecture.gif",
                    "https:\/\/slavlotski.com\/pictures\/execution_date-explain.jpg",
                    "https:\/\/slavlotski.com\/pictures\/trigger-rules.png",
                    "https:\/\/slavlotski.com\/pictures\/sensors-in-dag.png",
                    "https:\/\/slavlotski.com\/pictures\/failed-sensors-apache-airflow.png"
                ]
            }
        }
    ],
    "_e2_version": 4116,
    "_e2_ua_string": "Aegea 11.2 (v4116)"
}