Оркестратор задач — инструмент или система, предназначенные для управления и контроля выполнением задач в сложных вычислительных и информационных средах. Он облегчают процесс развертывания, автоматизации и управления выполнением задач, что позволяет повысить эффективность работы и оптимизировать ресурсы.

Одним из популярных оркестратором задач является Apache Airflow. Он, как и все инструменты, имеет свои преимущества и недостатки, о которых пойдет речь в данной статье

Apache Airflow

Apache Airflow - это платформа для программирования, планирования и мониторинга рабочих процессов (Workflows). Он позволяет создавать и управлять сложными зависимостями между задачами, автоматизировать их выполнение и мониторить процесс обработки данных.

Airflow предоставляет графический интерфейс для создания, выполнения и мониторинга рабочих процессов, также он позволяет создавать рабочие процессы в форме направленных ацикличных графов (DAG), где каждый узел представляет собой отдельную задачу. Это обеспечивает удобство и гибкость в управлении задачами и зависимостями между ними.

Airflow широко используется для автоматизации распределенных вычислений, обработки данных, управления ETL (Extract, Transform, Load) процессами, планирования задач и других сценариев работы с данными.

По умолчанию Apache Airflow использует SQLite, как базу данных.

Один из способов использования Apache Airflow - docker файл.

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

Начиная с версии 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 можно написать задачу, которая получает статус других задач. Данная задача будет полезна для:

1. Отчетность и мониторинг: Задача получения статуса задач может использоваться для создания мониторинговых или отчетных процедур, которые сообщают о статусе выполнения других задач. Например, вы можете хотеть получить уведомление или отправить отчет, когда определенная задача завершается успешно или неудачно.

2. Автоматизация бизнес‑процессов: Получение статуса задач также может быть использовано для автоматизации бизнес‑процессов. Например, если определенная задача завершилась успешно, вы можете автоматически запускать следующие действия или процессы.

3. Управление зависимостями: Получение статуса задач может помочь в управлении зависимостями и выполнением последующих действий в зависимости от статуса предыдущих задач. Например, если одна задача завершилась неудачно, вы можете хотеть принять решение о перезапуске другой задачи или выполнении альтернативных действий.

Пример кода с получением статуса задачи:

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 и конвейеры обработки данных»