Первый commit
This commit is contained in:
@@ -0,0 +1,286 @@
|
||||
# Основы построения DAG-файлов в Airflow
|
||||
|
||||
В предыдущем занятии вы познакомились с пользовательским интерфейсом Airflow и, вероятно, задались вопросами: как создать свой собственный DAG? Где писать код? Какие именно инструкции использовать? В этом уроке вы получите ответы на все эти вопросы, создадите свой первый DAG, изучите базовую структуру кода и сможете загрузить его в систему для запуска и наблюдения за результатами.
|
||||
|
||||
# Как устроен код DAG-файла
|
||||
|
||||
Как вы уже знаете из предыдущих уроков, DAG представляет собой граф вычислений, состоящий из отдельных задач (tasks). На уровне кода DAG — это обычный Python-файл, который описывает все задачи в рамках пайплайна и определяет последовательность их выполнения.
|
||||
|
||||
Любой DAG-файл состоит из нескольких ключевых компонентов:
|
||||
- Импорт необходимых модулей и библиотек
|
||||
- Настройка параметров и инициализация объекта DAG
|
||||
- Создание отдельных задач с помощью операторов
|
||||
- Определение порядка выполнения задач
|
||||
|
||||
Порядок этих компонентов имеет значение, поскольку Airflow — это Python-библиотека, и обращение к еще не инициализированным объектам приведет к ошибкам выполнения.
|
||||
|
||||
Давайте рассмотрим простой пример DAG-файла и разберем его по частям.
|
||||
|
||||
|
||||
*Пример структуры DAG-файла*
|
||||
|
||||
## Импорт необходимых модулей
|
||||
|
||||
Все начинается с подключения нужных библиотек и классов. Обязательно импортируйте класс DAG для создания графа вычислений. Затем подключайте операторы, которые понадобятся для описания ваших задач. Также могут потребоваться дополнительные функции или объекты для работы с данными.
|
||||
|
||||
```python
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from airflow import DAG
|
||||
from airflow.operators.bash import BashOperator
|
||||
from airflow.utils.dates import days_ago
|
||||
```
|
||||
|
||||
## Настройка параметров DAG
|
||||
|
||||
Следующим шагом идет объявление DAG с необходимыми параметрами:
|
||||
|
||||
В примере ниже создается DAG с идентификатором "sample_dag", который будет использовать общие параметры, определенные в словаре `default_args`. В данном случае, владелец DAG - это 'data_team', а дата начала выполнения - 1 января 2021 года.
|
||||
|
||||
```python
|
||||
default_args = {
|
||||
'start_date': datetime(2021, 1, 1),
|
||||
'owner': 'data_team'
|
||||
}
|
||||
|
||||
dag = DAG(
|
||||
"sample_dag",
|
||||
default_args=default_args,
|
||||
schedule_interval=None,
|
||||
)
|
||||
```
|
||||
|
||||
Обратите внимание на словарь `default_args`. Это стандартная практика для хранения общих параметров, которые будут применяться ко всем задачам в DAG. Такой подход помогает избежать дублирования кода и делает его более читаемым. В примере мы указываем владельца пайплайна и дату начала его работы.
|
||||
|
||||
Ключевыми обязательными параметрами являются `owner` и `start_date`. Без них DAG не сможет быть корректно инициализирован и не появится в пользовательском интерфейсе Airflow. Остальные параметры являются опциональными и добавляются по мере необходимости.
|
||||
|
||||
Для изучения всех доступных параметров рекомендуем обращаться к официальной документации Airflow.
|
||||
|
||||
## Создание задач с помощью операторов
|
||||
|
||||
После настройки DAG следует этап создания задач. Для каждого шага обработки данных создается отдельная переменная с соответствующим оператором.
|
||||
|
||||
В Airflow существует множество готовых операторов для различных задач:
|
||||
- `BashOperator` — для выполнения bash-команд
|
||||
- `PythonOperator` — для запуска Python-функций
|
||||
- Специализированные операторы для работы с популярными системами обработки данных (например, Apache Spark)
|
||||
|
||||
Каждая задача должна иметь уникальный идентификатор (`task_id`), который используется Airflow для отображения задачи в интерфейсе.
|
||||
|
||||
Пример создания простых задач:
|
||||
|
||||
В этом примере мы создаем две задачи с использованием BashOperator. Первая задача с идентификатором 'show_time' выводит текущее время, а вторая задача с идентификатором 'process_data' имитирует обработку данных с задержкой 3 секунды и возможностью повторного запуска при ошибках (до 2 раз).
|
||||
|
||||
```python
|
||||
# Задача для вывода текущего времени
|
||||
t1 = BashOperator(
|
||||
task_id='show_time',
|
||||
bash_command='date',
|
||||
dag=dag
|
||||
)
|
||||
|
||||
# Задача для имитации обработки с возможностью повторных попыток
|
||||
t2 = BashOperator(
|
||||
task_id='process_data',
|
||||
bash_command='sleep 3',
|
||||
retries=2,
|
||||
dag=dag
|
||||
)
|
||||
```
|
||||
|
||||
## Определение последовательности выполнения
|
||||
|
||||
Завершающий этап — указание порядка выполнения задач. В Airflow для этого используются стрелочные операторы:
|
||||
|
||||
В приведенном примере задача с идентификатором 'process_data' будет запускаться только после успешного завершения задачи 'show_time'.
|
||||
|
||||
```python
|
||||
t1 >> t2 # t2 запускается после завершения t1
|
||||
```
|
||||
|
||||
Альтернативный синтаксис:
|
||||
|
||||
Этот синтаксис эквивалентен предыдущему примеру, просто записан в обратном порядке. Задача 'show_time' должна завершиться перед запуском задачи 'process_data'.
|
||||
```python
|
||||
t2 << t1 # t1 должна завершиться перед запуском t2
|
||||
```
|
||||
|
||||
Можно также группировать задачи в списки для создания более сложных зависимостей:
|
||||
|
||||
В этом примере задачи 't2' и 't3' будут запускаться одновременно после завершения задачи 't1'. Затем задача 't4' запустится после завершения обеих задач 't2' и 't3'.
|
||||
|
||||
```python
|
||||
t1 >> [t2, t3] # t2 и t3 запускаются одновременно после t1
|
||||
[t2, t3] >> t4 # t4 запускается после завершения t2 и t3
|
||||
```
|
||||
|
||||
Для сложных пайплайнов можно создавать цепочки любой сложности:
|
||||
```python
|
||||
t1 >> [t2, t3] >> t4 >> [t5, t6, t7] >> t8
|
||||
```
|
||||
|
||||
## Полный пример DAG-файла
|
||||
|
||||
В этом полном примере мы создаем DAG с более подробной настройкой параметров. Обратите внимание на дополнительные параметры, такие как количество повторных попыток ('retries'), задержка между попытками ('retry_delay'), а также настройки уведомлений по электронной почте.
|
||||
|
||||
Вот как выглядит готовый DAG-файл в итоге:
|
||||
|
||||
```python
|
||||
from datetime import datetime, timedelta
|
||||
from airflow import DAG
|
||||
from airflow.operators.bash import BashOperator
|
||||
from airflow.utils.dates import days_ago
|
||||
|
||||
|
||||
default_args = {
|
||||
'owner': 'data_team',
|
||||
'depends_on_past': False,
|
||||
'start_date': days_ago(1),
|
||||
'email': ['data@example.com'],
|
||||
'email_on_failure': False,
|
||||
'email_on_retry': False,
|
||||
'retries': 2,
|
||||
'retry_delay': timedelta(minutes=3),
|
||||
}
|
||||
|
||||
dag = DAG(
|
||||
"sample_dag",
|
||||
default_args=default_args,
|
||||
schedule_interval=None,
|
||||
)
|
||||
|
||||
t1 = BashOperator(
|
||||
task_id='show_time',
|
||||
bash_command='date',
|
||||
dag=dag
|
||||
)
|
||||
|
||||
t2 = BashOperator(
|
||||
task_id='process_data',
|
||||
bash_command='sleep 3',
|
||||
retries=2,
|
||||
dag=dag
|
||||
)
|
||||
|
||||
t1 >> t2
|
||||
```
|
||||
|
||||
**Важное замечание**: код DAG-файла выполняется Airflow при каждом сканировании директории DAG. Поэтому в нем не следует размещать тяжелые вычисления, чтение больших файлов или запросы к базам данных. DAG-файл должен быть максимально легковесным. Сложные операции следует выносить в отдельные функции и вызывать их через соответствующие операторы.
|
||||
|
||||
💡 **Правило**: DAG-файл не должен содержать ресурсоемких операций, загрузки файлов или сложных вычислений.
|
||||
|
||||
# Создаем свой первый рабочий пайплайн
|
||||
|
||||
Теперь давайте создадим полноценный пайплайн для работы с реальными данными. В качестве примера мы будем использовать датасет с информацией о клиентах, который часто применяется в задачах анализа данных.
|
||||
|
||||
Наш пайплайн будет состоять из трех этапов:
|
||||
1. Загрузка исходного датасета из интернета
|
||||
2. Создание агрегированной таблицы по регионам и категориям
|
||||
3. Сохранение результата в базу данных PostgreSQL
|
||||
|
||||
Вот полный код нашего DAG:
|
||||
|
||||
```python
|
||||
import os
|
||||
import datetime as dt
|
||||
import pandas as pd
|
||||
from airflow.models import DAG
|
||||
from airflow.operators.python import PythonOperator
|
||||
from airflow.operators.bash import BashOperator
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
|
||||
|
||||
# Базовые параметры DAG
|
||||
args = {
|
||||
'owner': 'airflow',
|
||||
'start_date': dt.datetime(2020, 12, 23),
|
||||
'retries': 1,
|
||||
'retry_delay': dt.timedelta(minutes=1),
|
||||
}
|
||||
|
||||
def download_titanic_dataset():
|
||||
url = 'https://web.stanford.edu/class/archive/cs/cs109/cs109.1166/stuff/titanic.csv'
|
||||
df = pd.read_csv(url)
|
||||
engine = create_engine('postgresql+psycopg2://jovyan:jovyan@localhost:5432/de')
|
||||
df.to_sql('titanic', engine, index=False, if_exists='replace', schema='public')
|
||||
|
||||
|
||||
def pivot_dataset():
|
||||
engine = create_engine('postgresql+psycopg2://jovyan:jovyan@localhost:5432/de')
|
||||
titanic_df = pd.read_sql('select * from public.titanic', con=engine)
|
||||
|
||||
df = titanic_df.pivot_table(
|
||||
index=['Sex'],
|
||||
columns=['Pclass'],
|
||||
values='Name',
|
||||
aggfunc='count'
|
||||
).reset_index()
|
||||
|
||||
df.to_sql('titanic_pivot', engine, index=False, if_exists='replace', schema='public' )
|
||||
|
||||
dag = DAG(
|
||||
dag_id='titanic_pivot',
|
||||
schedule_interval=None,
|
||||
default_args=args,
|
||||
)
|
||||
|
||||
# Начальная задача для логирования
|
||||
start = BashOperator(
|
||||
task_id='start',
|
||||
bash_command='echo "Начинаем выполнение пайплайна! "',
|
||||
dag=dag,
|
||||
)
|
||||
|
||||
# Загрузка исходного датасета
|
||||
create_titanic_dataset = PythonOperator(
|
||||
task_id='download_titanic_dataset',
|
||||
python_callable=download_titanic_dataset,
|
||||
dag=dag,
|
||||
)
|
||||
|
||||
# Преобразование и сохранение сводной таблицы
|
||||
pivot_titanic_dataset = PythonOperator(
|
||||
task_id='pivot_dataset',
|
||||
python_callable=pivot_dataset,
|
||||
dag=dag,
|
||||
)
|
||||
|
||||
# Последовательность выполнения
|
||||
start >> create_titanic_dataset >> pivot_titanic_dataset
|
||||
```
|
||||
|
||||
Этот код использует `PythonOperator` для выполнения функций работы с данными, что является стандартной практикой для задач обработки данных. В примере функция `load_customer_dataset` загружает данные из внешнего источника, а `aggregate_customer_dataset` создает агрегированную таблицу.
|
||||
|
||||
## Загрузка DAG в учебную среду
|
||||
|
||||
Для тестирования нашего DAG в учебной среде выполните следующие шаги:
|
||||
|
||||
1. Сохраните код в файл с расширением `.py` (например, `customer_analysis_dag.py`)
|
||||
|
||||
Для тестирования нашего DAG в учебной среде выполните следующие шаги:
|
||||
|
||||
1. Сохраните код в файл с расширением `.py` (например, `customer_analysis_dag.py`)
|
||||
|
||||
2. Найдите запущенный контейнер с учебной средой Airflow:
|
||||
```bash
|
||||
docker ps
|
||||
```
|
||||
|
||||
3. Скопируйте файл в контейнер:
|
||||
```bash
|
||||
docker cp /путь/к/файлу/customer_analysis_dag.py [ID_КОНТЕЙНЕРА]:/lessons/dags/customer_analysis_dag.py
|
||||
```
|
||||
|
||||
Например:
|
||||
```bash
|
||||
docker cp ~/Desktop/customer_analysis_dag.py 4dc4fbfedcf4:/lessons/dags/customer_analysis_dag.py
|
||||
```
|
||||
|
||||
4. Подождите 30-60 секунд — Airflow автоматически обнаружит новый файл
|
||||
|
||||
5. Найдите ваш DAG в интерфейсе Airflow через строку поиска и запустите его
|
||||
|
||||
|
||||
|
||||
В этом уроке вы изучили основную структуру DAG-файлов, создали свой первый рабочий пайплайн, научились загружать его в учебную среду и запускать для получения результатов.
|
||||
Reference in New Issue
Block a user