🔒Открыть пост
и посмотреть медиа
и посмотреть медиа
Основное про Apache Airflow
Apache Airflow - платформа с довольно обширным функционалом, но в контексте аналитики данных используется по большей части как ETL-инструмент, а также как инструмент для автоматизации процессов через питон.
По сути, основная часть Airflow (особенно в дата-аналитике) - это python-код, который имеет некоторые различия с обычным питоном. Так как мы имеем дело с фреймворком, то мы можем пользоваться особыми классами (по типу DAG, BashOperator и т.п.). Также нужно учитывать, что при работе с Airflow мы не пишем то, как это должно выполняться. Тут используется более декларативный подход.
Среди этого "питоновского диалекта" главное, на что следует обратить внимание - это DAG. DAG (Directed Acyclic Graph) - это рабочий процесс, который состоит из задач и их зависимостей. В нем задачи выполняются в строгом порядке, а также нельзя зациклить выполнение этих задач. Основа, для чего существует DAG, выглядят так:
▶️Описать сами задачи, в чем они заключаются
▶️Задать порядок выполнения этих задач
▶️Задать периодичность выполнения всего этого процесса
Сам DAG создается так (пример):
with DAG(
# Базовые параметры
dag_id='my_sample_dag',
default_args=default_args,
description='Пример DAG',
start_date=datetime(2024, 1, 1),
# Расписание
schedule_interval='@daily',
# или cron-строка '0 3 * * *'
# Дополнительные настройки
catchup=False,
tags=['example', 'tutorial'],
max_active_runs=1
) as dag:
(перед этим можно задать аргументы, которые передадутся в DAG, например:)
default_args = {
'owner': 'admin',
'email': '[email protected]'
}
После этого пишутся задачи, в основном с использованием какого-либо оператора. Как пример:
process_data = PythonOperator(
task_id='process_data',
# Вызов python-функции
python_callable=python_function,
# передача контекста в функцию
provide_context=True
)
и определяется зависимость между задачами
start_task >> process_data >> end_task
Сами вышеупомянутые операторы бывают:
▶️BashOperator - для Bash-команд
▶️PythonOperator - для питона
▶️PostgresOperator - для Postgres
▶️DummyOperator - действий не делает, только "группирует" таски
▶️MySqlOperator - для MySQL
▶️BranchPythonOperator - для сложных Python-кодов с условиями (если... то...)
▶️BranchSQLOperator - для сложных SQL-запросов с условиями (если... то...)
▶️FileSensor - проверяет наличие файла в директории
и много других.
🟠В заключение как может в общих чертах выглядеть один из ETL-процессов в Airflow:
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
def extract():
...
def transform():
...
def load():
...
with DAG(
"my_etl_dag",
start_date=datetime(2025, 1, 1),
schedule_interval="@daily",
) as dag:
extract_task = PythonOperator(task_id="extract", python_callable=extract)
transform_task = PythonOperator(task_id="transform", python_callable=transform)
load_task = PythonOperator(task_id="load", python_callable=load)
extract_task >> transform_task >> load_task
Сперва как обычно импорт библиотек и написание типичных питоновских функций (в данном случае по ETL). Потом создаем DAG, который будет выполняться ежедневно, начиная с 01.01.25. Далее через PythonOperator задаем логику для каждого из 3 ETL тасков, а именно: задаем id и передаем в таски ранее написанные функции. В самом конце строим порядок для выполнения этих 3 задач.