company_banner

Airflow — инструмент, чтобы удобно и быстро разрабатывать и поддерживать batch-процессы обработки данных

    image


    Привет, Хабр! В этой статье я хочу рассказать об одном замечательном инструменте для разработки batch-процессов обработки данных, например, в инфраструктуре корпоративного DWH или вашего DataLake. Речь пойдет об Apache Airflow (далее Airflow). Он несправедливо обделен вниманием на Хабре, и в основной части я попытаюсь убедить вас в том, что как минимум на Airflow стоит смотреть при выборе планировщика для ваших ETL/ELT-процессов.


    Ранее я писал серию статей на тему DWH, когда работал в Тинькофф Банке. Теперь я стал частью команды Mail.Ru Group и занимаюсь развитием платформы для анализа данных на игровом направлении. Собственно, по мере появления новостей и интересных решений мы с командой будем рассказывать тут о нашей платформе для аналитики данных.


    Пролог


    Итак, начнем. Что такое Airflow? Это библиотека (ну или набор библиотек) для разработки, планирования и мониторинга рабочих процессов. Основная особенность Airflow: для описания (разработки) процессов используется код на языке Python. Отсюда вытекает масса преимуществ для организации вашего проекта и разработки: по сути, ваш (например) ETL-проект — это просто Python-проект, и вы можете его организовывать как вам удобно, учитывая особенности инфраструктуры, размер команды и другие требования. Инструментально всё просто. Используйте, например, PyCharm + Git. Это прекрасно и очень удобно!


    Теперь рассмотрим основные сущности Airflow. Поняв их суть и назначение, вы оптимально организуете архитектуру процессов. Пожалуй, основная сущность — это Directed Acyclic Graph (далее DAG).


    DAG


    DAG — это некоторое смысловое объединение ваших задач, которые вы хотите выполнить в строго определенной последовательности по определенному расписанию. Airflow представляет удобный web-интерфейс для работы с DAG’ами и другими сущностями:



    DAG может выглядеть таким образом:



    Разработчик, проектируя DAG, закладывает набор операторов, на которых будут построены задачи внутри DAG’а. Тут мы приходим еще к одной важной сущности: Airflow Operator.


    Операторы


    Оператор — это сущность, на основании которой создаются экземпляры заданий, где описывается, что будет происходить во время исполнения экземпляра задания. Релизы Airflow с GitHub уже содержат набор операторов, готовых к использованию. Примеры:


    • BashOperator — оператор для выполнения bash-команды.
    • PythonOperator — оператор для вызова Python-кода.
    • EmailOperator — оператор для отправки email’а.
    • HTTPOperator — оператор для работы с http-запросами.
    • SqlOperator — оператор для выполнения SQL-кода.
    • Sensor — оператор ожидания события (наступления нужного времени, появления требуемого файла, строки в базе БД, ответа из API — и т. д., и т. п.).

    Есть более специфические операторы: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator.


    Вы также можете разрабатывать операторы, ориентируясь на свои особенности, и использовать их в проекте. Например, мы создали MongoDBToHiveViaHdfsTransfer, оператор экспорта документов из MongoDB в Hive, и несколько операторов для работы с ClickHouse: CHLoadFromHiveOperator и CHTableLoaderOperator. По сути, как только в проекте возникает часто используемый код, построенный на базовых операторах, можно задуматься о том, чтобы собрать его в новый оператор. Это упростит дальнейшую разработку, и вы пополните свою библиотеку операторов в проекте.


    Далее все эти экземпляры задачек нужно выполнять, и теперь речь пойдет о планировщике.


    Планировщик


    Планировщик задач в Airflow построен на Celery. Celery — это Python-библиотека, позволяющая организовать очередь плюс асинхронное и распределенное исполнение задач. Со стороны Airflow все задачи делятся на пулы. Пулы создаются вручную. Как правило, их цель — ограничить нагрузку на работу с источником или типизировать задачи внутри DWH. Пулами можно управлять через web-интерфейс:



    Каждый пул имеет ограничение по количеству слотов. При создании DAG’а ему задается пул:


    ALERT_MAILS =  Variable.get("gv_mail_admin_dwh")
    DAG_NAME = 'dma_load'
    OWNER = 'Vasya Pupkin'
    DEPENDS_ON_PAST = True
    EMAIL_ON_FAILURE = True
    EMAIL_ON_RETRY = True
    RETRIES = int(Variable.get('gv_dag_retries'))
    POOL = 'dma_pool'
    PRIORITY_WEIGHT = 10
    
    start_dt = datetime.today() - timedelta(1)
    start_dt = datetime(start_dt.year, start_dt.month, start_dt.day)
    
    default_args = {
        'owner': OWNER,
        'depends_on_past': DEPENDS_ON_PAST,
        'start_date': start_dt,
        'email': ALERT_MAILS,
        'email_on_failure': EMAIL_ON_FAILURE,
        'email_on_retry': EMAIL_ON_RETRY,
        'retries': RETRIES,
        'pool': POOL,
        'priority_weight': PRIORITY_WEIGHT
    }
    dag = DAG(DAG_NAME, default_args=default_args)
    dag.doc_md = __doc__

    Пул, заданный на уровне DAG’а, можно переопределить на уровне задачи.
    За планировку всех задач в Airflow отвечает отдельный процесс — Scheduler. Собственно, Scheduler занимается всей механикой постановки задачек на исполнение. Задача, прежде чем попасть на исполнение, проходит несколько этапов:


    1. В DAG’е выполнены предыдущие задачи, новую можно поставить в очередь.
    2. Очередь сортируется в зависимости от приоритета задач (приоритетами тоже можно управлять), и, если в пуле есть свободный слот, задачу можно взять в работу.
    3. Если есть свободный worker celery, задача направляется в него; начинается работа, которую вы запрограммировали в задачке, используя тот или иной оператор.

    Достаточно просто.


    Scheduler работает на множестве всех DAG’ов и всех задач внутри DAG’ов.


    Чтобы Scheduler начал работу с DAG’ом, DAG’у нужно задать расписание:


    dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='@hourly')

    Есть набор готовых preset’ов: @once, @hourly, @daily, @weekly, @monthly, @yearly.


    Также можно использовать cron-выражения:


    dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='*/10 * * * *')

    Execution Date


    Чтобы разобраться в том, как работает Airflow, важно понимать, что такое Execution Date для DAG’а. В Airflow DAG имеет измерение Execution Date, т. е. в зависимости от расписания работы DAG’а создаются экземпляры задачек на каждую Execution Date. И за каждую Execution Date задачи можно выполнить повторно — или, например, DAG может работать одновременно в нескольких Execution Date. Это наглядно отображено здесь:



    К сожалению (а может быть, и к счастью: зависит от ситуации), если правится реализация задачки в DAG’е, то выполнение в предыдущих Execution Date пойдет уже с учетом корректировок. Это хорошо, если нужно пересчитать данные в прошлых периодах новым алгоритмом, но плохо, потому что теряется воспроизводимость результата (конечно, никто не мешает вернуть из Git’а нужную версию исходника и разово посчитать то, что нужно, так, как нужно).


    Генерация задач


    Реализация DAG’а — код на Python, поэтому у нас есть очень удобный способ сократить объем кода при работе, например, с шардированными источниками. Пускай у вас в качестве источника три шарда MySQL, вам нужно слазить в каждый и забрать какие-то данные. Причем независимо и параллельно. Код на Python в DAG’е может выглядеть так:


    connection_list = lv.get('connection_list')
    
    export_profiles_sql = '''
    SELECT
      id,
      user_id,
      nickname,
      gender,
      {{params.shard_id}} as shard_id
    FROM profiles
    '''
    
    for conn_id in connection_list:
        export_profiles = SqlToHiveViaHdfsTransfer(
            task_id='export_profiles_from_' + conn_id,
            sql=export_profiles_sql,
            hive_table='stg.profiles',
            overwrite=False,
            tmpdir='/data/tmp',
            conn_id=conn_id,
            params={'shard_id': conn_id[-1:], },
            compress=None,
            dag=dag
        )
        export_profiles.set_upstream(exec_truncate_stg)
        export_profiles.set_downstream(load_profiles)

    DAG получается таким:



    При этом можно добавить или убрать шард, просто скорректировав настройку и обновив DAG. Удобно!


    Можно использовать и более сложную генерацию кода, например работать с источниками в виде БД или описывать табличную структуру, алгоритм работы с таблицей и с учетом особенностей инфраструктуры DWH генерировать процесс загрузки N таблиц к вам в хранилище. Или же, например, работу с API, которое не поддерживает работу с параметром в виде списка, вы можете сгенерировать по этому списку N задач в DAG’е, ограничить параллельность запросов в API пулом и выгрести из API необходимые данные. Гибко!


    Репозиторий


    В Airflow есть свой бекенд-репозиторий, БД (может быть MySQL или Postgres, у нас Postgres), в которой хранятся состояния задач, DAG’ов, настройки соединений, глобальные переменные и т. д., и т. п. Здесь хотелось бы сказать, что репозиторий в Airflow очень простой (около 20 таблиц) и удобный, если вы хотите построить какой-либо свой процесс над ним. Вспоминается 100500 таблиц в репозитории Informatica, которые нужно было долго вкуривать, прежде чем понять, как построить запрос.


    Мониторинг


    Учитывая простоту репозитория, вы можете сами построить удобный для вас процесс мониторинга задачек. Мы используем блокнот в Zeppelin, где смотрим состояние задач:



    Это может быть и web-интерфейс самого Airflow:



    Код Airflow открыт, поэтому мы у себя добавили алертинг в Telegram. Каждый работающий инстанс задачи, если происходит ошибка, спамит в группу в Telegram, где состоит вся команда разработки и поддержки.


    Получаем через Telegram оперативное реагирование (если такое требуется), через Zeppelin — общую картину по задачам в Airflow.


    Итого


    Airflow в первую очередь open source, и не нужно ждать от него чудес. Будьте готовы потратить время и силы на то, чтобы выстроить работающее решение. Цель из разряда достижимых, поверьте, оно того стоит. Скорость разработки, гибкость, простота добавления новых процессов — вам понравится. Конечно, нужно уделять много внимания организации проекта, стабильности работы самого Airflow: чудес не бывает.


    Сейчас у нас Airflow ежедневно отрабатывает около 6,5 тысячи задач. По характеру они достаточно разные. Есть задачи загрузки данных в основное DWH из множества разных и очень специфических источников, есть задачи расчета витрин внутри основного DWH, есть задачи публикации данных в быстрое DWH, есть много-много разных задач — и Airflow все их пережевывает день за днем. Если же говорить цифрами, то это 2,3 тысячи ELT задач различной сложности внутри DWH (Hadoop), около 2,5 сотен баз данных источников, это команда из 4-ёх ETL разработчиков, которые делятся на ETL процессинг данных в DWH и на ELT процессинг данных внутри DWH и конечно ещё одного админа, который занимается инфраструктурой сервиса.


    Планы на будущее


    Количество процессов неизбежно растет, и основное, чем мы будем заниматься в части инфраструктуры Airflow, — это масштабирование. Мы хотим построить кластер Airflow, выделить пару ног для worker’ов Celery и сделать дублирующую себя голову с процессами планировки заданий и репозиторием.


    Эпилог


    Это, конечно, далеко не всё, что хотелось бы рассказать об Airflow, но основные моменты я постарался осветить. Аппетит приходит во время еды, попробуйте — и вам понравится :)

    Mail.Ru Group 736,67
    Строим Интернет
    Поделиться публикацией
    Похожие публикации
    Комментарии 24
    • +1
      Отличная штука для больших компаний! А какие решения вы бы предложили для компаний поменьше?
      • 0
        Спасибо за вопрос. Я хотел бы подчеркнуть что этот инструмент подходит для команда разных размеров и разных по объему задач. В силу того что он бесплатный, с открытым кодом, гибкий и удобный для быстрого старта вы можете организовать под свои требования свой сервис, будь у вас маленькая компания и вам будет достаточно одного сервера или же airflow подселенного на сервер БД, или будь у вас кластер Airflow для проворачивания тысяч задач и работы с гетерогенным хранилищем и гетерогенными источниками данных.
      • 0

        Очень хотелось бы услышать мнение автора о Spring dataflow в сравнении с airflow…

        • 0
          Со Spring dataflow не работал и мало что про него знаю. Если есть примеры и мнение по его части, напишите сюда, пожалуйста.
        • 0
          А можно какие-то конкретные примеры использования? Пока просто не очень понятно, для чего именно этот инструмент. Это что-то вроде Huginn (https://habrahabr.ru/post/330382/), или тут другие задачи?
          • 0
            Если я правильно понял из статьи про Huginn, то Huginn это такой self-service mix из ESB+BPM. Airflow это совсем не про это. Вы работали с хранилищами данных? Или наверняка слышали про λ архитектуру? Airflow про ETL/ELT и batch слой вашей λ-ы. Кейс может быть простой: вам нужно собрать по клиенту кучу данных из разных источников и например раз в день (можно чаще) обновлять его собирательный профиль.
          • 0
            если вы в амазоне, то swf. В принципе, все очень похоже
          • +1
            Для масштабирования не рассматривали интеграцию с Mesos?
            • 0
              Как делать масштабирование пока вопрос открытый. Оставлять CeleryExecutor или сделать шаг в сторону MesosExecutor — хз (хотел бы знать) как правильно. С Mesos экспертизы нет, а с Celery и Airflow работает у нас работает, и на Celery ещё пытаемся одну внутреннюю штуку писать. По этому сейчас кажется что несколько сельдереев (worker node) для нас будет оптимальным решением.
              Если делать масштабирование Airflow на Mesos, в чем преимущество?
          • 0
            Юр, привет! Пара, возможно, тупых вопросов:

            1. Если у меня есть Informatica, то мне вот этот тул что-то даст? Помимо более лаконичного репозитория, возможности интеграции с telegram и прочих глобально не очень важных фич?
            Возможно, его стоит использовать параллельно для какого-то пула задач со схожими характеристиками?
            2. Я правильно понял, что у вас сейчас Airflow — Это основное ETL/ELT средство и других не держите? Во всяком случае у тебя на направлении? Правильно понял, что в принципе у вас стратегия использовать по возможности только опенсёрс?
            3. Как с глобальной поддержкой? Были кейсы, которые сами не смогли решить и пришлось к кому-то обращаться? Если да, то к кому обращались?

            • 0
              О, интерпрайзные парни подоспели :) Привет, Вов!

              1. Если у тебя информатика и ты плотно на ней сидишь, то не думаю что тебе что-то даст Airflow. Разве что если только поможет открыть глаза на что более светлое ;) Про стоимость конечно глупо говорить. Про скорость разработки и добавления на среду новых процессов, вот про про что нужно говорить. Про кодогенерацию, про сравнительную простоту масштабирования. Про возможность делать кастом операторы стоит говорить.
              2. У нас в хранилище для батч уровня да, только Airflow. Есть такая стратегия, сейчас всё хранилище строится на опенсорс.
              3. Кейсы ты имеешь ввиду что? Падений? Или требования которые было непонятно как решить на текущей сборке Airflow? Падения были, разрешали сами, + стековерфло, +гитхаб. Требования которае не могли бы решить, тут как мне кажется если задача адекватная, и источник в меру адекватный и алгоритм в меру адекватный, то для airflow проблем не будет, т.к. есть PythonOperator на котором в целом можно решить достаточно заковыристые задачи по получению данных. Ты можешь мне подкинуть задачу, я могу тебе на шагах расписать как это будет реализовано на Airflow.
            • 0
              Apache nifi не пробовали для ваших задач? Чем не понравился?
              • 0
                Для батч уровня нет. Для стримминговой части системы (которая сейчас на совсем начальной стадии) можно рассмотреть.
                • 0
                  Попробовал, понравилось. Спасибо за наводку!
              • 0

                Интересно, сколько времени занимает создание такого количества заданий у команды из 4 человек? Сколько по времени длилась разработка?

                • 0
                  Хороший вопрос. Начали в сентябре прошлого года (2016). У нас было пустое хранилище и ноль дагов на среде. Было некое наследие. Наследие мигрировало (с необходимыми переработками) на новую инфраструктуру, это где-то 1/4 текущих процессов, остальное было разработано считай в течение года.
                • 0
                  Интересно. Мы в Badoo сделали очень похожую систему, только на php. Некоторые детали отличаются, но общая картина абсолютно такая же.

                  Все процессы разделены на цепочки (chains). Внутри одной цепочки задачи выполняются всегда последовательно. Цепочки сами выполняют всегда параллельно, но с учётом зависимостей. Схема просто непобедимая в своей простоте.

                  Как дела обстоят с логикой повторных попыток в случае ошибки (retry)? Как работает перезагрузка блока старых данных? Например, хотим перезагрузить все логи транзакций за 2016 год и всё, что от них зависит за то же самое время. Удобно ли это сделать?

                  Как организован devel environment? Можно ли целиком протестировать ETL на своей ветке, используя sample от production данных и не мешая при этом другим разработчикам?
                  • +1
                    Схема просто непобедимая в своей простоте.

                    Вот вообще сложно поспорить :)

                    Как дела обстоят с логикой повторных попыток в случае ошибки (retry)?


                    Кол-во и интервал ритраев можно настроить на уровне дага, можно переопределить на уровне задачек. Есть параметр оповещать на ритрай, есть на финальный феилд. Достаточно просто и удобно.

                    Как работает перезагрузка блока старых данных? Например, хотим перезагрузить все логи транзакций за 2016 год и всё, что от них зависит за то же самое время. Удобно ли это сделать?


                    Если даг построен корректно, т.е. с применением макросов airflow и за определенный день (дату) даг работает с определенными данными, то достаточно просто найти в запусках дага требуемый день и с него почистить статусы задачек в этом дне и в последующих днях (делается это легко). Планировщик обработает эти задачки ещё раз. Так же есть процедура backfill (https://airflow.apache.org/cli.html), можно с помощь неё перезапустить определенный период в даге.

                    Как организован devel environment? Можно ли целиком протестировать ETL на своей ветке, используя sample от production данных и не мешая при этом другим разработчикам?


                    Эта часть у нас пока плохо организована. Здесь ещё предстоит много работы. Нашему DWH всего год. Куда двигаться в целом понятно: у каждого разраба свои репы, система пул реквестов, скрипт копирования прод таблиц/данных в песочницу для разработчиков (в рамках одного hadoop кластера).
                    • 0
                      Блеск.

                      Видимо, последний вопрос. Можно ли сделать ожидание на некий внешний источник за пределами Airflow? Например, «появились файлы в HDFS» или «внешний API ответил, что он готов». В теории, можно сделать через retry и долго пытаться, пока не получится. Но ещё лучше совсем в очередь задачу не ставить, пока данные не готовы. Можно ли так сделать?

                      > Здесь ещё предстоит много работы.

                      Поделюсь немного информацией о том, какой подход у нас успешно работает. Возможно, какие-то идеи пригодятся.

                      • У каждого разработчика свой отдельный девел на отдельном домене. Конфиги цепочек хранятся в JSON файлах и отслеживаются гитом. Конфиги — не в базе.
                      • База с очередью задач одна на всех, но у каждой таблицы есть колонка «user». Каждый разработчик видит только свою очередь и свои логи. Есть кнопка «Debug reset», которая очищает все данные разработчика и позволяет начать тест с чистого листа.
                      • Колоночная СУБД (Exasol) одна на всех, но есть возможность указать custom prefix для имён схем. Например, обычная схема «FINANCE», а с префиксом будет «VASYA_FINANCE». Такие схемы создаются автоматически на основании DDL из кода. Вместо имён схем в коде везде используются плейсхолдеры. Имена таблиц остаются без изменений. Это позволяет избежать конфликтов DDL и локов, когда сразу несколько человек тестируют что-то одновременно.
                      • При отладке можно указать конкретные цепочки (DAG'и), которые нужно протестировать. Зависимости учитываются автоматически и тоже выполняются. Помогает побыстрее тестировать небольшие изменения и делать code review за разумное время.
                      • На девел загружаются живые данные с production источников, но применяется жёсткий sampling. Вместо миллиарда записей тянем сто записей. Вместо загрузки всех файлов тянем по сто рядов из первых и последних файлов. Всё это позволяет провести некий аналог функционального теста на весь ETL за 20-30 минут.


                      В идеале, стремимся к полностью изолированным девелам. Насколько это упрощает жизнь — словами не передать.
                      • +1
                        Видимо, последний вопрос. Можно ли сделать ожидание на некий внешний источник за пределами Airflow? Например, «появились файлы в HDFS» или «внешний API ответил, что он готов». В теории, можно сделать через retry и долго пытаться, пока не получится. Но ещё лучше совсем в очередь задачу не ставить, пока данные не готовы. Можно ли так сделать?


                        Для таких задач есть отличный набор операторов — sensor-ы. Они как раз про, что бы что-нибудь подождать. Есть и минус, минус в том что сенсоры это такие же задачи airflow, требующие постановки в очередь и свободного слота и воркера. Мы держим сенсоры в отдельном пуле, тем самым контролируем их кол-во и приоритезируем их.

                        Спасибо за информацию по подходу к организации среды для разработки.
                  • 0
                    А как быстро у вас планировщик от одной завершенной задачки к следующей переходит?
                    • 0
                      Хороший вопрос. Переходы между задачками, их запуск — это не сильная сторона Airflow. Есть много накладных расходов, таких как постановка задачек в очередь, их приоритезиция при запуске, в случае наличия свободного слота в пуле и свободного воркера сельдерея. Планирощик работает постоянно и постоянно пытается толкать задачи, делает он с настраиваемым интервалом (scheduler_heartbeat_sec). У нас при не загруженном сервере, т.е. при наличие свободных слотов в пуле и свободных воркеров сельдерея последующие задачки запускаются в интервале 5-10 секунд.

                    Только полноправные пользователи могут оставлять комментарии. Войдите, пожалуйста.

                    Самое читаемое