На обложке: Воздушный поток (air flow) вокруг самолета, источник: fineartamerica.com
В предыдущих статьях я разобрала, как наладить поток логов из конструктора Dialogflow в BigQuery, и теперь почва для построения отчётов построена. Сегодня покажу, как настроить один DAG – исполняемую по расписанию группу команд в Airflow. Я исхожу из предпосылки, что вы уже развернули программу на своем сервере / устройстве и имеете доступ к UI. Эта статья подходит тем, кто решил автоматизировать в своей работе то, что достаточно отлажено, чтобы поддаться автоматизации. В моем случае это скрипты, рассчитывающие ежемесячную эффективность одного из ботов по конечным состояниям диалогов.
Это статья из цикла «5 ETL для зоопарка ботов». В нём я пошагово разбираю, как наладить потоки данных из разных библиотек и конструкторов ботов на разных языках и стеках. В основе лежат Python и его библиотеки. Вот предыдущие статьи цикла, подводящие к текущей:
DAG (Directed Acyclic Graph – направленный ациклический граф) – просто коллекция различных задач. Вероятно, вы уже сталкивались с термином «граф», если проходили курсы по машинному обучению. В контексте Airflow же слово имеет примерно такое же значение – это поток задач, который может разветвляться:
Потому модуль DAG импортируется всегда, чтобы связать различные задачи воедино.
Для начала я импортирую необходимые библиотеки:
from datetime import datetime, timedelta from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.telegram.operators.telegram import TelegramOperator from airflow.utils.trigger_rule import TriggerRule
BashOperator нужен, чтобы запускать силами bash-скрипты, лежащие по соседству. Классы datetime, timedelta модуля datetime нужны, чтобы описать поведение графа при ошибке. TriggerRule понадобится, чтобы уведомлять меня в Telegram только при ошибке.
Почти все рабочие чаты у меня живут в Telegram, потому удобно получать уведомления об исполнении в специальный чат, где живут Airflow-бот и его братья по другим проектам. Поэтому я импортировала TelegramOperator.
Если вы не знаете, как создать бота в Telegram, то используйте раздел «Создание бота» из вот этого гайда. Кратко: передав команду /start боту @BotFather, мы следуем инструкции, называем бота, задаем ему идентификатор для поиска и в конце получаем токен:
Скопировав токен, создаём функцию, уведомляющую об исполнении задач через мессенджер:
send_telegram_message = TelegramOperator( task_id=»send_telegram_message», token=»<Токен бота в Telegram>», chat_id=»Идентификатор чата, куда добавлен бот», text=»Расчет сводки выполнен.» )
Чтобы получить chat_id, используйте бота @RawDataBot. Токен бота оставляем в одноименном поле.
Теперь создаём DAG и описываем некоторые обязательные характеристики:
with DAG( «daily_effectiveness», # Уникальное имя графа, отобразится в консоли default_args={ «depends_on_past»: False, # Зависимость задач от предыдущих «retries»: 1, # Число перепопыток в случае неудаче «retry_delay»: timedelta(seconds=30) # Интервал между попытками }, description=»Ежемесячная сводка маркетплейса», Описание, появится в консоли при наведении на название DAG’а schedule_interval=’@monthly’, # Ежемесячное исполнение start_date=datetime(2023, 7, 1), # Когда начать исполнение по расписанию catchup=False, tags=[«Маркетплейс», «Dialogflow», «BigQuery»], ) as dag: Ключевая особенность Airflow заключается в том, что запуск DAG по расписанию – нечто вроде копирования группы задач для каждого нужного времени. Планировщик (scheduler – его мы запускали в прошлой статье) проверяет, не работает ли до сих пор предыдущая копия графа. Эта концепция называется catchup (можно перевести как «подхват»).
Теперь создаём задачи (tasks) – отдельные исполняемые команды в рамках графа. В моём случае для расчёта сводки эффективности необходимо предварительно войти в виртуальную среду airflow_env, и только потом исполнять подпрограмму dialogflow-to-bigquery.py силами bash:
t1 = BashOperator( task_id=»entering_virtual_environment», # Идентификатор таски для отслеживания в консоли bash_command=»source /home/fitwist/airflow/airflow_env/bin/activate», retries=2 # ) t2 = BashOperator( task_id=»calculating_marketplace_effectiveness», depends_on_past=False, bash_command=»python3 /home/fitwist/airflow/df-to-looker/dialogflow-to-bigquery.py», retries=2 )
Задаем последовательность задач:
t1 >> t2 >> t3 >> t4 >> TelegramOperator( task_id=»send_telegram_message», token=»6191376785:AAHf9thqeAuaIjou3DjzEFiI06bYY9FDKlI», chat_id=»-1001200247335″, trigger_rule=TriggerRule.ONE_FAILED, text=»Группа чат-бота: одна из ежедневных выгрузок не выполнена. Проверь логи (Recent Tasks / Failed / Task Id / Log).» )
И вуаля! Ваш первый DAG готов! Когда у меня впервые получилось запустить Airflow, чувство было, словно в космос полетела, ведь такая технология в резюме очень котируется у дата-аналитиков.
Data Scientist Открытие, Москва, можно удалённо, По итогам собеседования tproger.ru Вакансии на tproger.ru
Полностью DAG будет выглядеть так:
from datetime import datetime, timedelta from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.telegram.operators.telegram import TelegramOperator send_telegram_message = TelegramOperator( task_id=»send_telegram_message», token=»<Токен бота в Telegram>», chat_id=»Идентификатор чата, куда добавлен бот», text=»Расчет сводки выполнен.» ) with DAG( «daily_effectiveness», # Идентификатор, отобразится в консоли default_args={ «depends_on_past»: False, # Зависимость задач от предыдущих «retries»: 1, # Число перепопыток в случае неудаче «retry_delay»: timedelta(seconds=30) # Интервал между попытками }, description=»Ежемесячная сводка маркетплейса», Описание, появится в консоли при наведении на название DAG schedule_interval=’@monthly’, # Ежемесячное исполнение start_date=datetime(2023, 7, 1), # Когда начать исполнение по расписанию catchup=False, tags=[«Маркетплейс», «Dialogflow», «BigQuery»], ) as dag: t1 = BashOperator( task_id=»entering_virtual_environment», # Идентификатор таски для отслеживания в консоли bash_command=»source /home/fitwist/airflow/airflow_env/bin/activate», retries=2 # ) t2 = BashOperator( task_id=»calculating_marketplace_effectiveness», depends_on_past=False, bash_command=»python3 /home/fitwist/airflow/df-to-looker/dialogflow-to-bigquery.py», retries=2 ) t1 >> t2 >> t3 >> t4 >> TelegramOperator( task_id=»send_telegram_message», token=»6191376785:AAHf9thqeAuaIjou3DjzEFiI06bYY9FDKlI», chat_id=»-1001200247335″, trigger_rule=TriggerRule.ONE_FAILED, text=»Группа чат-бота: одна из ежедневных выгрузок не выполнена. Проверь логи (Recent Tasks / Failed / Task Id / Log).» )
Можно даже получать уведомления об ошибке любым удобным способом. Мне удобно в Telegram, потому использую триггер ONE_FAILED, то есть “когда хотя бы одна задача не выполнена”.
Этот и другие сниппеты из цикла статей можно найти в моем репозитории на GitHub.
Airflow – это настоящий швейцарский нож. В нём, как в Python, написана утилита почти под любую потребность. И на покрытие тредами Stack Overflow грех жаловаться.
Когда я только делала первые шаги в направлении автоматизации, то существенную моральную помощь также оказывала прекрасная документация: там и примеры, и грамотно дозированные страницы. Даже база графов на любой вкус на GitHub. Так и хочется создателям за неё выдать награду.
Это уникальный в своем роде опенсорсный продукт, который позволяет даже новичкам предъявить полезнейший функционал – автоматизацию чего угодно. Он не требует знания теории вероятности и основ машинного обучения, однако при наличии Airflow в вашем резюме позволяет требовать 100К+.
В следующей статье мы посмотрим, как выглядит сводка эффективности, которую мы теперь можем визуализировать силами Google Looker (экс-Data Studio).
В предыдущих статьях я разобрала, как наладить поток логов из конструктора Dialogflow в BigQuery, и теперь почва для построения отчётов построена. Сегодня покажу, как настроить один DAG – исполняемую по расписанию группу команд в Airflow. Я исхожу из предпосылки, что вы уже развернули программу на своем сервере / устройстве и имеете доступ к UI. Эта статья подходит тем, кто решил автоматизировать в своей работе то, что достаточно отлажено, чтобы поддаться автоматизации. В моем случае это скрипты, рассчитывающие ежемесячную эффективность одного из ботов по конечным состояниям диалогов.
Это статья из цикла «5 ETL для зоопарка ботов». В нём я пошагово разбираю, как наладить потоки данных из разных библиотек и конструкторов ботов на разных языках и стеках. В основе лежат Python и его библиотеки. Вот предыдущие статьи цикла, подводящие к текущей:
- Анонс цикла с перечнем технологий
- Настройка потока логов «Из Dialogflow в BigQuery»
- Python для аналитики ad hoc из BigQuery
- Развертывание Airflow
Что такое DAG
DAG (Directed Acyclic Graph – направленный ациклический граф) – просто коллекция различных задач. Вероятно, вы уже сталкивались с термином «граф», если проходили курсы по машинному обучению. В контексте Airflow же слово имеет примерно такое же значение – это поток задач, который может разветвляться:
Потому модуль DAG импортируется всегда, чтобы связать различные задачи воедино.
Вернемся к настройке
Для начала я импортирую необходимые библиотеки:
from datetime import datetime, timedelta from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.telegram.operators.telegram import TelegramOperator from airflow.utils.trigger_rule import TriggerRule
BashOperator нужен, чтобы запускать силами bash-скрипты, лежащие по соседству. Классы datetime, timedelta модуля datetime нужны, чтобы описать поведение графа при ошибке. TriggerRule понадобится, чтобы уведомлять меня в Telegram только при ошибке.
Подключение бота Telegram
Почти все рабочие чаты у меня живут в Telegram, потому удобно получать уведомления об исполнении в специальный чат, где живут Airflow-бот и его братья по другим проектам. Поэтому я импортировала TelegramOperator.
Если вы не знаете, как создать бота в Telegram, то используйте раздел «Создание бота» из вот этого гайда. Кратко: передав команду /start боту @BotFather, мы следуем инструкции, называем бота, задаем ему идентификатор для поиска и в конце получаем токен:
Скопировав токен, создаём функцию, уведомляющую об исполнении задач через мессенджер:
send_telegram_message = TelegramOperator( task_id=»send_telegram_message», token=»<Токен бота в Telegram>», chat_id=»Идентификатор чата, куда добавлен бот», text=»Расчет сводки выполнен.» )
Чтобы получить chat_id, используйте бота @RawDataBot. Токен бота оставляем в одноименном поле.
Граф
Теперь создаём DAG и описываем некоторые обязательные характеристики:
with DAG( «daily_effectiveness», # Уникальное имя графа, отобразится в консоли default_args={ «depends_on_past»: False, # Зависимость задач от предыдущих «retries»: 1, # Число перепопыток в случае неудаче «retry_delay»: timedelta(seconds=30) # Интервал между попытками }, description=»Ежемесячная сводка маркетплейса», Описание, появится в консоли при наведении на название DAG’а schedule_interval=’@monthly’, # Ежемесячное исполнение start_date=datetime(2023, 7, 1), # Когда начать исполнение по расписанию catchup=False, tags=[«Маркетплейс», «Dialogflow», «BigQuery»], ) as dag: Ключевая особенность Airflow заключается в том, что запуск DAG по расписанию – нечто вроде копирования группы задач для каждого нужного времени. Планировщик (scheduler – его мы запускали в прошлой статье) проверяет, не работает ли до сих пор предыдущая копия графа. Эта концепция называется catchup (можно перевести как «подхват»).
Задачи
Теперь создаём задачи (tasks) – отдельные исполняемые команды в рамках графа. В моём случае для расчёта сводки эффективности необходимо предварительно войти в виртуальную среду airflow_env, и только потом исполнять подпрограмму dialogflow-to-bigquery.py силами bash:
t1 = BashOperator( task_id=»entering_virtual_environment», # Идентификатор таски для отслеживания в консоли bash_command=»source /home/fitwist/airflow/airflow_env/bin/activate», retries=2 # ) t2 = BashOperator( task_id=»calculating_marketplace_effectiveness», depends_on_past=False, bash_command=»python3 /home/fitwist/airflow/df-to-looker/dialogflow-to-bigquery.py», retries=2 )
Задаем последовательность задач:
t1 >> t2 >> t3 >> t4 >> TelegramOperator( task_id=»send_telegram_message», token=»6191376785:AAHf9thqeAuaIjou3DjzEFiI06bYY9FDKlI», chat_id=»-1001200247335″, trigger_rule=TriggerRule.ONE_FAILED, text=»Группа чат-бота: одна из ежедневных выгрузок не выполнена. Проверь логи (Recent Tasks / Failed / Task Id / Log).» )
И вуаля! Ваш первый DAG готов! Когда у меня впервые получилось запустить Airflow, чувство было, словно в космос полетела, ведь такая технология в резюме очень котируется у дата-аналитиков.
Data Scientist Открытие, Москва, можно удалённо, По итогам собеседования tproger.ru Вакансии на tproger.ru
Полностью DAG будет выглядеть так:
from datetime import datetime, timedelta from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.telegram.operators.telegram import TelegramOperator send_telegram_message = TelegramOperator( task_id=»send_telegram_message», token=»<Токен бота в Telegram>», chat_id=»Идентификатор чата, куда добавлен бот», text=»Расчет сводки выполнен.» ) with DAG( «daily_effectiveness», # Идентификатор, отобразится в консоли default_args={ «depends_on_past»: False, # Зависимость задач от предыдущих «retries»: 1, # Число перепопыток в случае неудаче «retry_delay»: timedelta(seconds=30) # Интервал между попытками }, description=»Ежемесячная сводка маркетплейса», Описание, появится в консоли при наведении на название DAG schedule_interval=’@monthly’, # Ежемесячное исполнение start_date=datetime(2023, 7, 1), # Когда начать исполнение по расписанию catchup=False, tags=[«Маркетплейс», «Dialogflow», «BigQuery»], ) as dag: t1 = BashOperator( task_id=»entering_virtual_environment», # Идентификатор таски для отслеживания в консоли bash_command=»source /home/fitwist/airflow/airflow_env/bin/activate», retries=2 # ) t2 = BashOperator( task_id=»calculating_marketplace_effectiveness», depends_on_past=False, bash_command=»python3 /home/fitwist/airflow/df-to-looker/dialogflow-to-bigquery.py», retries=2 ) t1 >> t2 >> t3 >> t4 >> TelegramOperator( task_id=»send_telegram_message», token=»6191376785:AAHf9thqeAuaIjou3DjzEFiI06bYY9FDKlI», chat_id=»-1001200247335″, trigger_rule=TriggerRule.ONE_FAILED, text=»Группа чат-бота: одна из ежедневных выгрузок не выполнена. Проверь логи (Recent Tasks / Failed / Task Id / Log).» )
Можно даже получать уведомления об ошибке любым удобным способом. Мне удобно в Telegram, потому использую триггер ONE_FAILED, то есть “когда хотя бы одна задача не выполнена”.
Этот и другие сниппеты из цикла статей можно найти в моем репозитории на GitHub.
Заключение
Airflow – это настоящий швейцарский нож. В нём, как в Python, написана утилита почти под любую потребность. И на покрытие тредами Stack Overflow грех жаловаться.
Когда я только делала первые шаги в направлении автоматизации, то существенную моральную помощь также оказывала прекрасная документация: там и примеры, и грамотно дозированные страницы. Даже база графов на любой вкус на GitHub. Так и хочется создателям за неё выдать награду.
Это уникальный в своем роде опенсорсный продукт, который позволяет даже новичкам предъявить полезнейший функционал – автоматизацию чего угодно. Он не требует знания теории вероятности и основ машинного обучения, однако при наличии Airflow в вашем резюме позволяет требовать 100К+.
В следующей статье мы посмотрим, как выглядит сводка эффективности, которую мы теперь можем визуализировать силами Google Looker (экс-Data Studio).
