Apache Airflow: преимущества и недостатки
Оркестратор задач — инструмент или система, предназначенные для управления и контроля выполнением задач в сложных вычислительных и информационных средах. Он облегчают процесс развертывания, автоматизации и управления выполнением задач, что позволяет повысить эффективность работы и оптимизировать ресурсы.
Одним из популярных оркестратором задач является Apache Airflow. Он, как и все инструменты, имеет свои преимущества и недостатки, о которых пойдет речь в данной статье
Apache Airflow
Apache Airflow - это платформа для программирования, планирования и мониторинга рабочих процессов (Workflows). Он позволяет создавать и управлять сложными зависимостями между задачами, автоматизировать их выполнение и мониторить процесс обработки данных.
Airflow предоставляет графический интерфейс для создания, выполнения и мониторинга рабочих процессов, также он позволяет создавать рабочие процессы в форме направленных ацикличных графов (DAG), где каждый узел представляет собой отдельную задачу. Это обеспечивает удобство и гибкость в управлении задачами и зависимостями между ними.
Airflow широко используется для автоматизации распределенных вычислений, обработки данных, управления ETL (Extract, Transform, Load) процессами, планирования задач и других сценариев работы с данными.
По умолчанию Apache Airflow использует SQLite, как базу данных.
Один из способов использования Apache Airflow - docker файл.
docker-compose
version: '3.8'
x-airflow-common:
&airflow-common
image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.7.3}
environment:
&airflow-common-env
AIRFLOW__CORE__EXECUTOR: LocalExecutor
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
AIRFLOW__CORE__FERNET_KEY: ''
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
AIRFLOW__API__AUTH_BACKENDS: 'airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session'
AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK: 'true'
_PIP_ADDITIONAL_REQUIREMENTS: ${_PIP_ADDITIONAL_REQUIREMENTS:-}
volumes:
- ${AIRFLOW_PROJ_DIR:-.}/dags:/opt/airflow/dags
- ${AIRFLOW_PROJ_DIR:-.}/logs:/opt/airflow/logs
- ${AIRFLOW_PROJ_DIR:-.}/config:/opt/airflow/config
- ${AIRFLOW_PROJ_DIR:-.}/plugins:/opt/airflow/plugins
user: "${AIRFLOW_UID:-50000}:0"
depends_on:
&airflow-common-depends-on
postgres:
condition: service_healthy
services:
postgres:
image: postgres:13
environment:
POSTGRES_USER: airflow
POSTGRES_PASSWORD: airflow
POSTGRES_DB: airflow
volumes:
- postgres-db-volume:/var/lib/postgresql/data
healthcheck:
test: ["CMD", "pg_isready", "-U", "airflow"]
interval: 10s
retries: 5
start_period: 5s
restart: always
airflow-webserver:
<<: *airflow-common
command: webserver
ports:
- "8080:8080"
healthcheck:
test: ["CMD", "curl", "--fail", "http://localhost:8080/health"]
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-scheduler:
<<: *airflow-common
command: scheduler
healthcheck:
test: ["CMD", "curl", "--fail", "http://localhost:8974/health"]
interval: 30s
timeout: 10s
retries: 5
start_period: 30s
restart: always
depends_on:
<<: *airflow-common-depends-on
airflow-init:
condition: service_completed_successfully
airflow-init:
<<: *airflow-common
entrypoint: /bin/bash
# yamllint disable rule:line-length
command:
- -c
- |
function ver() {
printf "%04d%04d%04d%04d" $${1//./ }
}
airflow_version=$$(AIRFLOW__LOGGING__LOGGING_LEVEL=INFO && gosu airflow airflow version)
airflow_version_comparable=$$(ver $${airflow_version})
min_airflow_version=2.2.0
min_airflow_version_comparable=$$(ver $${min_airflow_version})
if (( airflow_version_comparable < min_airflow_version_comparable )); then
echo
echo -e "\033[1;31mERROR!!!: Too old Airflow version $${airflow_version}!\e[0m"
echo "The minimum Airflow version supported: $${min_airflow_version}. Only use this or higher!"
echo
exit 1
fi
if [[ -z "${AIRFLOW_UID}" ]]; then
echo
echo -e "\033[1;33mWARNING!!!: AIRFLOW_UID not set!\e[0m"
echo "If you are on Linux, you SHOULD follow the instructions below to set "
echo "AIRFLOW_UID environment variable, otherwise files will be owned by root."
echo "For other operating systems you can get rid of the warning with manually created .env file:"
echo " See: https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html#setting-the-right-airflow-user"
echo
fi
one_meg=1048576
mem_available=$$(($$(getconf _PHYS_PAGES) * $$(getconf PAGE_SIZE) / one_meg))
cpus_available=$$(grep -cE 'cpu[0-9]+' /proc/stat)
disk_available=$$(df / | tail -1 | awk '{print $$4}')
warning_resources="false"
if (( mem_available < 4000 )) ; then
echo
echo -e "\033[1;33mWARNING!!!: Not enough memory available for Docker.\e[0m"
echo "At least 4GB of memory required. You have $$(numfmt --to iec $$((mem_available * one_meg)))"
echo
warning_resources="true"
fi
if (( cpus_available < 2 )); then
echo
echo -e "\033[1;33mWARNING!!!: Not enough CPUS available for Docker.\e[0m"
echo "At least 2 CPUs recommended. You have $${cpus_available}"
echo
warning_resources="true"
fi
if (( disk_available < one_meg * 10 )); then
echo
echo -e "\033[1;33mWARNING!!!: Not enough Disk space available for Docker.\e[0m"
echo "At least 10 GBs recommended. You have $$(numfmt --to iec $$((disk_available * 1024 )))"
echo
warning_resources="true"
fi
if [[ $${warning_resources} == "true" ]]; then
echo
echo -e "\033[1;33mWARNING!!!: You have not enough resources to run Airflow (see above)!\e[0m"
echo "Please follow the instructions to increase amount of resources available:"
echo " https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html#before-you-begin"
echo
fi
mkdir -p /sources/logs /sources/dags /sources/plugins
chown -R "${AIRFLOW_UID}:0" /sources/{logs,dags,plugins}
exec /entrypoint airflow version
environment:
<<: *airflow-common-env
_AIRFLOW_DB_MIGRATE: 'true'
_AIRFLOW_WWW_USER_CREATE: 'true'
_AIRFLOW_WWW_USER_USERNAME: ${_AIRFLOW_WWW_USER_USERNAME:-airflow}
_AIRFLOW_WWW_USER_PASSWORD: ${_AIRFLOW_WWW_USER_PASSWORD:-qwerty12345!}
_PIP_ADDITIONAL_REQUIREMENTS: ''
user: "0:0"
volumes:
- ${AIRFLOW_PROJ_DIR:-.}:/sources
airflow-cli:
<<: *airflow-common
profiles:
- debug
environment:
<<: *airflow-common-env
CONNECTION_CHECK_MAX_COUNT: "0"
command:
- bash
- -c
- airflow
volumes:
postgres-db-volume:
Несколько популярных операторов в Apache Airflow:
EmptyOperator
Этот оператор представляет собой очень простой оператор в Apache Airflow, который не выполняет никаких реальных действий.
- Ожидание или задержка: можно использовать EmptyOperator для создания задержки между выполнением задач в вашем DAG.
- Создание структуры DAG: может быть использован для создания структуры вашего DAG, когда у вас еще нет конкретных операций для некоторых частей рабочего процесса.
- Использование в качестве заполнителя: может понадобиться создать «место действия» для определенных зависимостей
- Тестирование: при разработке и тестировании DAG, вы можете использовать EmptyOperator, чтобы вставить временную заглушку вместо реальных операций.
Пример кода с EmptyOperator:
from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.empty import EmptyOperator
default_my_args = {
'owner': 'PB_Academy',
'retries': 5
}
with DAG (
dag_id = 'dag_empty',
description='ДАГ с использованием bash оператора',
schedule_interval='@once',
default_args= default_my_args,
start_date=days_ago(1)
) as dag:
start_task = EmptyOperator(
task_id = 'start_task'
)
task_1_1 = EmptyOperator(
task_id = 'task_1_1'
)
task_1_2 = EmptyOperator(
task_id = 'task_1_2'
)
union_task = EmptyOperator(
task_id = 'union'
)
task_2_1 = EmptyOperator(
task_id = 'task_2_1'
)
task_2_2 = EmptyOperator(
task_id = 'task_2_2'
)
finish_task = EmptyOperator(
task_id = 'finish_task'
)
start_task >> [task_1_1, task_1_2] >> union_task >>[task_2_1, task_2_2] >> finish_task
Отображение dag с помощью графа
Начиная с версии 2.3.0 Apache Airflow появился данный оператор. До этого существовал DummyOperator. В версии 2.3.4 разработчики призывают переходить на EmptyOperator.
BashOperator
Этот оператор используется для выполнения произвольных bash‑команд или скриптов. Он является универсальным инструментом для выполнения различных командных операций в рамках ваших рабочих процессов.
Пример кода с BashOperator:
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
default_my_args = {
'owner': 'PB_Academy',
'retries': 5,
'retry_delay': timedelta(minutes=1)
}
with DAG(
dag_id='dag_bush',
default_args= default_my_args,
description='ДАГ с использованием bash оператора',
start_date= datetime(2024, 5, 7, 16, 50, 0),
schedule_interval='@daily'
) as dag:
task1 = BashOperator(
task_id = 'first_task',
bash_command= 'mkdir PB_Academy'
)
task2 = BashOperator(
task_id = 'second_task',
bash_command= 'touch tmp.txt'
)
task3 = BashOperator(
task_id = 'thrird_task',
bash_command= 'echo Hello World'
)
task1 >> task2 >> task3
PostgresOperator
Данный оператор позволяет выполнять SQL-запросы в базе данных PostgreSQL. Он предоставляет удобный способ выполнения различных операций с базой данных PostgreSQL в рамках ваших DAG.
Для того, чтобы установить PostgresOperator рекомендуем выполнить следующую команду:
pip install apache-airflow-providers-postgres
Пример кода с PostgresOperator:
from airflow import DAG
from datetime import datetime, timedelta
from airflow.providers.postgres.operators.postgres import PostgresOperator
default_my_args = {
'owner': 'PB_Academy',
'retries': 5
}
with DAG(
dag_id = 'dag_postgres',
schedule_interval='@once',
default_args= default_my_args,
start_date= datetime(2023, 12, 8)
) as dag:
task_pg_create = PostgresOperator(
task_id='task_pg_create',
postgres_conn_id='postgres_default',
sql='CREATE TABLE test_1(id SERIAL, date TIMESTAMP);'
)
task_pg_insert = PostgresOperator(
task_id='task_pg_insert',
postgres_conn_id='postgres_default',
sql='INSERT INTO test_1 (date) VALUES (localtimestamp);'
)
task_pg_create >> task_pg_insert
PythonOperator
Этот оператор позволяет запускать произвольные функции Python в вашем DAG. Это может быть полезно, если вам нужно выполнить какой‑то кастомный код или манипуляции с данными в Python.
Для общения между задачами в рамках одного dag существует механизм — XCom. Это словарь, из которого можно брать данные. Как конфиги, так и результаты других задач. Данные в XCom могут быть как public, так и private. Push — отправить данные в словарь, pull — получить. Стоит добавить, что XCom не подчищает за собой данные, так что нужно побеспокоиться, чтобы после выполнения сессии dag удалял историю.
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
default_my_args = {
'owner': 'PB_Academy',
'retries': 5,
'retry_delay': timedelta(minutes=5)
}
def print_param(ti):
sql = ti.xcom_pull(task_ids = 'get_param', key = 'SQL')
nosql = ti.xcom_pull(task_ids = 'get_param', key = 'NoSQL')
print(f"SQL database: {sql} \n NoSQL database: {nosql}")
def get_param(ti):
ti.xcom_push(key = 'SQL', value = ['MS SQL','PostgreSQL','Oracle'])
ti.xcom_push(key = 'NoSQL', value = ['MongoDB','Redis','Elasticsearch'])
with DAG(
dag_id='dag_python',
default_args= default_my_args,
description='-',
start_date= datetime(2023, 12, 8, 8,47),
schedule_interval='@once'
) as dag:
task1 = PythonOperator(
task_id='task_print_param',
python_callable = print_param,
)
task2 = PythonOperator(
task_id='task_get_param',
python_callable = get_param
)
task2 >> task1
Task Flow API
Task Flow API — это часть Apache Airflow, которая предоставляет возможность программно создавать и управлять задачами и их зависимостями в DAG. Он позволяет создавать и изменять DAG и его структуру изнутри самого Python‑кода, а не только из конфигурационных файлов.
Пример кода с иcпользованием Task Flow API:
from datetime import datetime
from airflow.decorators import dag, task
default_my_args = {
'owner': 'PB_Academy',
'retries': 5
}
@dag(
dag_id = 'taskflow_api_dag',
default_args = default_my_args,
start_date = datetime(2023,12,10),
schedule_interval='@daily'
)
def calculate_etl():
@task(multiple_outputs = True)
def get_numbers():
return {
'number_1' : 21,
'number_2' : 99
}
@task
def calculate(number_1, number_2):
print(number_1 + number_2)
dicts = get_numbers()
calculate(dicts['number_1'], dicts['number_2'])
my_dag = calculate_etl()
Дополнительно
catchup — параметр в Apache Airflow, который определяет, должны ли старые даты выполнения задач быть выполнены при первом запуске DAG. Когда catchup установлен в True, Apache Airflow автоматически запускает пропущенные даты выполнения задач после того, как DAG был включен в работу.
Пример кода с catch up:
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime
default_my_args = {
'owner': 'PB_Academy',
'retries': 5
}
with DAG(
dag_id='dag_catchup',
default_args= default_my_args,
start_date= datetime(2023, 12, 7, 16, 50, 0),
schedule_interval='@daily',
catchup=True
) as dag:
task1 = BashOperator(
task_id = 'first_task',
bash_command= 'echo Hello World'
)
В Apache Airflow можно написать задачу, которая получает статус других задач. Данная задача будет полезна для:
- Отчетность и мониторинг: Задача получения статуса задач может использоваться для создания мониторинговых или отчетных процедур, которые сообщают о статусе выполнения других задач. Например, вы можете хотеть получить уведомление или отправить отчет, когда определенная задача завершается успешно или неудачно.
- Автоматизация бизнес‑процессов: Получение статуса задач также может быть использовано для автоматизации бизнес‑процессов. Например, если определенная задача завершилась успешно, вы можете автоматически запускать следующие действия или процессы.
- Управление зависимостями: Получение статуса задач может помочь в управлении зависимостями и выполнением последующих действий в зависимости от статуса предыдущих задач. Например, если одна задача завершилась неудачно, вы можете хотеть принять решение о перезапуске другой задачи или выполнении альтернативных действий.
Пример кода с получением статуса задачи:
from airflow.models.dagrun import DagRun
from airflow.decorators import dag, task
from datetime import datetime
default_my_args = {
'owner': 'PB_Academy',
'retries': 5
}
@dag(
dag_id = 'dag_with_status',
default_args = default_my_args,
start_date = datetime(2023,12,10),
schedule_interval='@daily'
)
def calculate_etl():
@task()
def get_status_dag():
dag_id = 'dag_with_status'
dag_runs = DagRun.find(dag_id=dag_id)
print(f'Length: {len(dag_runs)}')
for dag_run in dag_runs:
if dag_run.external_trigger == False :
print(f'Dag Name {dag_run}\n Dag Status {dag_run.state}')
@task()
def success_task():
4 / 2
@task()
def fail_task():
4 / 0
success_task()
fail_task()
get_status_dag()
my_dag = calculate_etl()
Параметр schedule_interval используется для определения расписания выполнения задач в рамках DAG. Этот параметр указывает, с какой периодичностью DAG и его задачи должны выполняться.schedule_interval может быть определен различными способами, включая крон-выражения, объекты datetime.timedelta или ключевые слова, такие как @daily, @hourly и т. д.
Вывод
Таким образом, Apache Airflow часто выбирают для сложных рабочих процессов с большим количеством задач. Подводя итог, рассмотрим основные плюсы и минусы данной системы:
|
Apache Airflow |
|
|
Плюсы |
Минусы |
|
Гибкость Airflow написан на Python, это делает его гибким и легко интегрируемым с другими инструментами. |
Сложность настройки Наличие неявных зависимостей может усложнить начальную настройку и конфигурацию. |
|
Мощный пользовательский интерфейс Графический веб-интерфейс позволяет легко создавать и мониторить рабочие процессы. |
Требования к ресурсам Для управления большими рабочими процессами требуются значительные вычислительные ресурсы. |
|
Расширяемость Поддерживает множество плагинов и интеграций с различными сервисами и базами данных. |
|
|
Масштабируемость Модульная архитектура и очереди сообщений позволяют обрабатывать большое количество задач. |
|
Для изучения данной темы рекомендуем прочитать книгу Баса Харенслака и Джулиана де Руйтера «Apache Airflow и конвейеры обработки данных»
