150 lines
8.6 KiB
Markdown
150 lines
8.6 KiB
Markdown
# Понятное введение в ключевые понятия Airflow
|
|
|
|
В этом материале мы разберем фундаментальные концепции Apache Airflow, которые вам обязательно понадобятся при создании пайплайнов обработки данных. Знание этих базовых терминов и компонентов поможет вам эффективно работать с системой.
|
|
|
|
# **Что такое DAG и почему он важен**
|
|
|
|
Сердце Apache Airflow — это DAG (Directed Acyclic Graph), что переводится как "направленный ациклический граф". Давайте разберем это сложное название по частям:
|
|
|
|
- **Граф** — это структура, где элементы (узлы) связаны между собой стрелками (ребрами)
|
|
- **Направленный** — означает, что связи имеют четкое направление: от одного элемента к другому, как последовательные этапы в обработке данных
|
|
- **Ациклический** — гарантирует, что вы не можете вернуться к уже пройденному узлу, избегая бесконечных циклов
|
|
|
|
Проще говоря, DAG — это упорядоченный набор задач, которые выполняются строго по расписанию и никогда не повторяются в рамках одного запуска. Создавая пайплайны для обработки данных, вы фактически создаете DAG.
|
|
|
|
```mermaid
|
|
flowchart LR
|
|
n1((1)) --> n2((2))
|
|
n1 --> n3((3))
|
|
|
|
n2 --> n4((4))
|
|
n3 --> n6((6))
|
|
n6 --> n7((7))
|
|
|
|
n7 --> n12
|
|
n4 --> n12((12))
|
|
```
|
|
|
|
# Шаги вашего пайплайна: Задачи
|
|
|
|
Задачи — это отдельные шаги в вашем процессе обработки данных. Каждая задача представляет собой конкретную бизнес-операцию:
|
|
- Проверка наличия файла в хранилище
|
|
- Выполнение SQL-запроса
|
|
- Запуск Python-функции с бизнес-логикой
|
|
- Отправка уведомления
|
|
|
|
💡 **Совет по проектированию**: Разбивайте сложные процессы на мелкие, атомарные задачи. Такой подход делает пайплайн более гибким, упрощает тестирование и помогает быстрее находить и исправлять ошибки.
|
|
|
|
На примере ниже видно, как задача E зависит от успешного завершения всех предыдущих задач:
|
|
|
|
```mermaid
|
|
flowchart LR
|
|
A["Task A"] --> B["Task B"]
|
|
A --> C["Task C"]
|
|
|
|
B --> D["Task D"]
|
|
C --> D
|
|
|
|
D --> E["Task E"]
|
|
```
|
|
|
|
Airflow позволяет создавать сложные сценарии:
|
|
- Зависимости между разными DAG-ами с помощью TriggerDagRunOperator и ExternalTaskSensor
|
|
- Условное выполнение задач в зависимости от результатов предыдущих шагов
|
|
- Сложные ветвления и параллельные ветки выполнения
|
|
|
|
```mermaid
|
|
flowchart LR
|
|
%% без цветов, только форма и рамка
|
|
classDef taskGroup stroke-width:3px;
|
|
classDef endpoint stroke-width:2px,stroke-dasharray: 5 3;
|
|
|
|
start(("начало")):::endpoint --> print_start["print_start_bash"]
|
|
|
|
print_start --> py1["python_function_with_input_1"]
|
|
print_start --> py4["python_function_with_input_4"]
|
|
|
|
py1 --> py2["python_function_with_input_2"]
|
|
py1 --> py3["python_function_with_input_3"]
|
|
|
|
py4 --> py5["python_function_with_input_5"]
|
|
py4 --> py6["python_function_with_input_6"]
|
|
|
|
py2 --> py_join["python_function_with_input"]
|
|
py3 --> py_join
|
|
py5 --> py_join
|
|
py6 --> py_join
|
|
|
|
py_join --> tg["task_group_with_two_tasks"]:::taskGroup
|
|
|
|
tg --> dyn123["dynamic_task_123"]
|
|
tg --> dyn456["dynamic_task_456"]
|
|
|
|
dyn123 --> finish(("конец")):::endpoint
|
|
dyn456 --> finish
|
|
```
|
|
|
|
# Инструменты для выполнения: Операторы
|
|
|
|
Операторы — это готовые шаблоны, которые определяют, КАК выполнять ваши задачи. Думайте о них как о строительных блоках для вашего пайплайна.
|
|
|
|
Пример простого DAG с двумя задачами:
|
|
|
|
В этом примере создается DAG с идентификатором 'greeting_dag', который запускается каждые 10 минут. DAG включает в себя две задачи: 'start_process' (использует DummyOperator как точку старта) и 'greeting_task' (использует PythonOperator для выполнения функции приветствия).
|
|
|
|
```python
|
|
from airflow import DAG
|
|
from airflow.operators.dummy import DummyOperator
|
|
from airflow.operators.python import PythonOperator
|
|
from datetime import datetime
|
|
|
|
def greet_user():
|
|
print('Welcome to Airflow!')
|
|
|
|
dag = DAG('greeting_dag',
|
|
description='Simple Greeting DAG',
|
|
schedule_interval='*/10 * * * *',
|
|
start_date=datetime(2023, 1, 1),
|
|
catchup=False)
|
|
|
|
start_task = DummyOperator(task_id='start_process', retries=2)
|
|
greeting_task = PythonOperator(task_id='greeting_task', python_callable=greet_user)
|
|
|
|
start_task >> greeting_task
|
|
```
|
|
|
|
В этом примере задача `greeting_task` использует `PythonOperator` для запуска функции `greet_user`, которая выводит приветственное сообщение в логи.
|
|
|
|
Airflow предоставляет множество готовых операторов:
|
|
- **PythonOperator** — выполнение Python-кода
|
|
- **BashOperator** — запуск команд и скриптов в терминале
|
|
- **PostgresOperator** — выполнение SQL-запросов (аналоги есть для MySQL, Oracle, Hive)
|
|
- **EmailOperator** — отправка электронных писем
|
|
- **DummyOperator** — "заглушка" для организации структуры пайплайна
|
|
|
|
💡 **Важное различие**: Задача определяет ЧТО нужно сделать (бизнес-логика), а оператор — КАК это сделать (техническая реализация).
|
|
|
|
# Ожидание событий: Сенсоры
|
|
|
|
Сенсоры — это специальный тип операторов, предназначенный для ожидания определенных условий или событий. Они идеально подходят для событийно-ориентированных пайплайнов.
|
|
|
|
Популярные сенсоры включают:
|
|
- **PythonSensor** — ожидает, пока функция не вернет `True`
|
|
- **S3Sensor** — проверяет наличие файла в S3-бакете
|
|
- **RedisPubSubSensor** — ждет поступления сообщения в очередь
|
|
- **RedisKeySensor** — проверяет существование ключа в Redis
|
|
|
|
Airflow поддерживает расширение функциональности через providers и позволяет создавать собственные операторы и сенсоры под специфические нужды.
|
|
|
|
# Краткое резюме
|
|
|
|
Давайте закрепим ключевые понятия:
|
|
|
|
- **DAG** — ваш пайплайн обработки данных, объединяющий задачи в логическую последовательность
|
|
- **Задача** — отдельный шаг в пайплайне, представляющий конкретную бизнес-операцию
|
|
- **Оператор** — технический инструмент для выполнения задачи (Python, SQL, Bash и т.д.)
|
|
- **Сенсор** — специальный оператор для ожидания внешних событий или условий
|
|
|
|
Взаимосвязь этих компонентов можно представить так:
|
|
|
|
 |