diff --git a/02 - Понятное введение в ключевые понятия Airflow.md b/02 - Понятное введение в ключевые понятия Airflow.md index ccac530..3050d96 100644 --- a/02 - Понятное введение в ключевые понятия Airflow.md +++ b/02 - Понятное введение в ключевые понятия Airflow.md @@ -107,12 +107,14 @@ dag = DAG('greeting_dag', 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 = DummyOperator(task_id='start_process', retries=2, dag=dag) +greeting_task = PythonOperator(task_id='greeting_task', python_callable=greet_user, dag=dag) start_task >> greeting_task ``` +Параметр `dag=dag` связывает каждый оператор с созданным DAG. Если объявить задачи внутри `with DAG(...) as dag:`, привязка произойдет автоматически. Стрелка `>>` задает порядок выполнения задач. + В этом примере задача `greeting_task` использует `PythonOperator` для запуска функции `greet_user`, которая выводит приветственное сообщение в логи. Airflow предоставляет множество готовых операторов: @@ -157,4 +159,4 @@ flowchart LR Task1 -.->|Как делать?| Op1["Оператор\n(например, PythonOperator)"] Task2 -.->|Чего ждать?| Sen1["Сенсор\n(например, S3Sensor)"] Task3 -.->|Как делать?| Op2["Оператор\n(например, PostgresOperator)"] -``` \ No newline at end of file +``` diff --git a/07 - Шаблоны, переменные и подключения в Airflow.md b/07 - Шаблоны, переменные и подключения в Airflow.md index 618fc87..5eb5226 100644 --- a/07 - Шаблоны, переменные и подключения в Airflow.md +++ b/07 - Шаблоны, переменные и подключения в Airflow.md @@ -8,45 +8,53 @@ Синтаксис шаблонов интуитивно понятен — переменная заключается в двойные фигурные скобки с пробелами по краям. Для эффективного использования достаточно знать, какое значение возвращает конкретный шаблон, и разместить его в нужном месте кода. -Подробный справочник доступных шаблонов можно найти в [официальной документации Airflow](https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html#variables). +Подробный справочник доступных шаблонов можно найти в [документации Airflow 2.9.2](https://airflow.apache.org/docs/apache-airflow/2.9.2/templates-ref.html#variables). -Вот практические примеры использования: +Каждый пример ниже - отдельный DAG. Для запуска сохраните один блок кода в файл `.py` внутри `airflow-docker/dags/`. Затем найдите DAG в UI и запустите его вручную. **Передача даты выполнения в задачу:** -В этом примере создается DAG с идентификатором "template_example", который запускается каждые 15 минут. Задача "display_date" использует шаблон {{ ds }} для вывода текущей даты выполнения в формате YYYY-MM-DD. +В этом примере DAG `template_example` запускается каждые 15 минут. Задача `display_date` выводит логическую дату запуска в формате YYYY-MM-DD через шаблон `{{ ds }}`. ```python -... +from datetime import datetime + +from airflow import DAG +from airflow.operators.bash import BashOperator dag = DAG( dag_id="template_example", schedule_interval="*/15 * * * *", - default_args=default_args -) + start_date=datetime(2023, 1, 1), + catchup=False, +) -t1 = BashOperator(task_id="display_date", bash_command="echo {{ ds }}") - -t1 +t1 = BashOperator(task_id="display_date", bash_command="echo {{ ds }}", dag=dag) ``` **Использование шаблонов для лучшей отслеживаемости задач:** В этом примере создается DAG с идентификатором "template_tracking_example", который запускается каждые 20 минут. Вторая задача использует шаблон {{ ds }} в команде bash, чтобы явно указывать дату обработки в логах. ```python -... +from datetime import datetime + +from airflow import DAG +from airflow.operators.bash import BashOperator dag = DAG( dag_id="template_tracking_example", schedule_interval="*/20 * * * *", - default_args=default_args -) + start_date=datetime(2023, 1, 1), + catchup=False, +) -t1 = BashOperator(task_id="show_date", bash_command="echo {{ ds }}") -t2 = BashOperator(task_id="process_for_date", bash_command="echo Processing for {{ ds }}") +t1 = BashOperator(task_id="show_date", bash_command="echo {{ ds }}", dag=dag) +t2 = BashOperator(task_id="process_for_date", bash_command="echo Processing for {{ ds }}", dag=dag) t1 >> t2 ``` +В первом DAG одна задача, во втором - две связанные задачи. В Details выбранной задачи откройте Rendered Templates: вместо `{{ ds }}` будет дата. Та же дата появится в логе после запуска. Привязку операторов к DAG здесь задает `dag=dag`. + # Безопасные переменные (Variables) Переменные Airflow представляют собой пары "ключ-значение", хранящиеся в метадатабазе системы. Они идеально подходят для хранения конфигурационных параметров, таких как пути к скриптам, имена таблиц или другие настройки, которые должны быть доступны в разных DAG. @@ -134,69 +142,62 @@ task = BashOperator( - Системы уведомлений (Email, Telegram) - И многие другие через Airflow Providers -При использовании операторов, взаимодействующих с внешними системами, Airflow автоматически ищет соответствующее подключение с суффиксом `_default`. Однако можно явно указать альтернативное подключение: +Оператор получает настройки по идентификатору подключения. Например, `PostgresOperator` использует параметр `postgres_conn_id`. PostgreSQL provider уже установлен в учебном стенде. -В приведенном примере используется PostgresOperator для создания таблицы в базе данных. Вместо использования подключения по умолчанию, явно указывается подключение с идентификатором 'my_postgres_conn'. - -```python -from airflow.providers.postgres.operators.postgres import PostgresOperator - -create_table = PostgresOperator( - task_id='create_user_table', - sql=''' - CREATE TABLE users( - user_id integer NOT NULL, - created_at TIMESTAMP NOT NULL - );''', - postgres_conn_id='my_postgres_conn' -) -``` - -Управление подключениями доступно через интерфейс Airflow (Admin → Connections). Если требуемый тип подключения отсутствует, его можно добавить установкой соответствующего Airflow Provider из [официального репозитория](https://airflow.apache.org/docs/#providers-packages-docs-apache-airflow-providers-index-html). - -Управление подключениями доступно через интерфейс Airflow (раздел **Admin → Connections**). Подключения хранятся в метадатабазе Airflow, а пароли и другие чувствительные поля шифруются с помощью Fernet и маскируются в UI и логах. +Подключения хранятся в базе метаданных Airflow. Пароли и другие чувствительные поля шифруются с помощью Fernet и маскируются в UI и логах. ### Как создать подключение к PostgreSQL через UI -Интерфейс ниже соответствует Airflow 2.9.x и стандартному Docker-стенду из документации: +Используйте учебную БД из [README стенда](airflow-docker/README.md#запуск). В контейнере Airflow она доступна по имени сервиса `postgres-training`; адрес `localhost:5432` предназначен для подключения с вашего компьютера. -1. Откройте веб-интерфейс Airflow. -2. В верхнем меню выберите **Admin → Connections**. -3. В правом верхнем углу нажмите кнопку **+ Add a new record**. -4. В форме укажите параметры: +1. Откройте **Admin > Connections** и нажмите **+ Add a new record**. +2. Заполните форму: - - **Connection Id**: `my_postgres_conn` - Это имя мы будем использовать в коде DAG (параметр `postgres_conn_id`). + - **Connection Id**: `my_postgres_conn`. Это имя будет указано в коде DAG. - **Connection Type**: `Postgres` - - **Host**: `postgres` - (так называется контейнер PostgreSQL в типовом `docker-compose.yaml` из официальной инструкции). - - **Schema**: `airflow` - (имя базы данных; в учебном стенде можно использовать стандартную БД). - - **Login**: `airflow` - - **Password**: `airflow` + - **Host**: `postgres-training` + - **Database / Schema**: `training` + - **Login**: `student` + - **Password**: `student` - **Port**: `5432` -5. Нажмите кнопку **Test** (если доступна) — Airflow попробует подключиться к базе. -6. Если тест успешен, нажмите **Save**. +3. Нажмите **Save**. Кнопка **Test** может быть отключена в конфигурации Airflow; подключение можно проверить запуском DAG ниже. -Теперь подключение с идентификатором `my_postgres_conn` доступно во всех DAG’ах. В примере ниже PostgresOperator явно использует это подключение: +База `airflow` на сервисе `postgres-metadata` хранит служебные данные Airflow. Для учебных таблиц используйте `training`. + +Сохраните пример в `airflow-docker/dags/connection_example_dag.py` и запустите `connection_example` через UI: ```python +from datetime import datetime + +from airflow import DAG from airflow.providers.postgres.operators.postgres import PostgresOperator -create_table = PostgresOperator( - task_id="create_user_table", - sql=""" - CREATE TABLE users( - user_id INTEGER NOT NULL, - created_at TIMESTAMP NOT NULL - ); - """, - postgres_conn_id="my_postgres_conn", -) +with DAG( + dag_id="connection_example", + start_date=datetime(2023, 1, 1), + schedule=None, + catchup=False, +) as dag: + create_table = PostgresOperator( + task_id="create_user_table", + sql=""" + CREATE TABLE IF NOT EXISTS public.connection_demo_users ( + user_id INTEGER NOT NULL, + created_at TIMESTAMP NOT NULL + ); + """, + postgres_conn_id="my_postgres_conn", + ) ``` -Для успешной работы с внешними системами сначала необходимо создать соответствующее подключение, а затем использовать его идентификатор в операторах вашего DAG. +Задача должна завершиться успешно. В учебной БД появится пустая таблица `public.connection_demo_users`; повторный запуск сохранит ее. Проверить наличие таблицы можно из SQL-консоли стенда: + +```sql +SELECT to_regclass('public.connection_demo_users'); +``` + +Результат - `connection_demo_users`. Если задача упала, откройте ее лог и сверьте Conn Id в коде с именем подключения, затем Host и остальные поля формы. Правка настроек действует при следующем выполнении задачи. Более подробную информацию о настройке подключений можно найти в [документации Airflow](https://airflow.apache.org/docs/apache-airflow/2.9.3/howto/connection.html). diff --git a/09 - Пулы, XCom, TaskGroup и алертинг в Airflow.md b/09 - Пулы, XCom, TaskGroup и алертинг в Airflow.md index 5cd97b1..7f1f58b 100644 --- a/09 - Пулы, XCom, TaskGroup и алертинг в Airflow.md +++ b/09 - Пулы, XCom, TaskGroup и алертинг в Airflow.md @@ -2,7 +2,7 @@ В этом материале мы познакомимся с управлением ресурсами через пулы задач, обменом данными между задачами через XCom, группировкой задач с помощью TaskGroup и настройкой системы оповещений (алертинга) в Airflow. -Начнем с улучшения нашего базового пайплайна — проведем рефакторинг кода для лучшей читаемости и поддержки. +Опыты с этими возможностями собраны в [блоке 8 практики](educational-tasks.md#resources); callback и ошибки разобраны в [блоке 7](educational-tasks.md#errors). ## Управление ресурсами с помощью пулов задач @@ -151,72 +151,62 @@ XCom упрощает взаимодействие между задачами ## Группировка задач с помощью TaskGroup -Для удобства визуализации задачи можно группировать в веб-интерфейсе Airflow (начиная с версии 2.0). Повторяющиеся или логически связанные задачи можно объединить в группы: +TaskGroup объединяет задачи в сворачиваемую группу на графе. Зависимости и состояния остаются у отдельных задач. Группировку задают в Python-коде; UI позволяет раскрыть ее и посмотреть содержимое. -В этом примере задачи сгруппированы в три логические группы: 'data_extraction', 'data_transformation' и 'data_loading'. Каждая группа содержит несколько задач, которые выполняются последовательно внутри группы. Затем группы связаны между собой, чтобы показать общий порядок выполнения этапов обработки данных. +В примере три группы. Внутри каждой задачи идут последовательно. Группы извлечения и преобразования независимы, а загрузка ждет завершения обеих. `EmptyOperator` ничего не обрабатывает: здесь мы смотрим только на устройство графа. ```python -with TaskGroup("data_extraction") as extraction_group: - extract_1 = DummyOperator(task_id="extract_source_1") - extract_2 = DummyOperator(task_id="extract_source_2") - extract_3 = DummyOperator(task_id="extract_source_3") - extract_1 >> extract_2 >> extract_3 +from datetime import datetime -with TaskGroup("data_transformation") as transformation_group: - transform_1 = DummyOperator(task_id="transform_step_1") - transform_2 = DummyOperator(task_id="transform_step_2") - transform_1 >> transform_2 +from airflow import DAG +from airflow.operators.empty import EmptyOperator +from airflow.utils.task_group import TaskGroup -with TaskGroup("data_loading") as loading_group: - load_1 = DummyOperator(task_id="load_to_target_1") - load_2 = DummyOperator(task_id="load_to_target_2") - load_1 >> load_2 +with DAG( + dag_id="task_group_example", + start_date=datetime(2023, 1, 1), + schedule=None, + catchup=False, +) as dag: + with TaskGroup("data_extraction") as extraction_group: + extract_1 = EmptyOperator(task_id="extract_source_1") + extract_2 = EmptyOperator(task_id="extract_source_2") + extract_3 = EmptyOperator(task_id="extract_source_3") + extract_1 >> extract_2 >> extract_3 -[extraction_group, transformation_group] >> loading_group + with TaskGroup("data_transformation") as transformation_group: + transform_1 = EmptyOperator(task_id="transform_step_1") + transform_2 = EmptyOperator(task_id="transform_step_2") + transform_1 >> transform_2 + + with TaskGroup("data_loading") as loading_group: + load_1 = EmptyOperator(task_id="load_to_target_1") + load_2 = EmptyOperator(task_id="load_to_target_2") + load_1 >> load_2 + + [extraction_group, transformation_group] >> loading_group ``` -Обратите внимание: при использовании TaskGroup последовательность задач указывается внутри группы после объявления всех задач, а в конце DAG описывается последовательность выполнения самих групп. +Здесь TaskGroup создается внутри `with DAG(...)`, поэтому группа и ее задачи получают один DAG. При отдельном объявлении `dag = DAG(...)` группу нужно привязать явно: `TaskGroup("data_extraction", dag=dag)`. -Ниже приведена упрощённая схема зависимостей между тремя группами задач: +Схема повторяет зависимости из кода: ```mermaid flowchart LR - - %% group1 - subgraph G1["group1"] - g1_t1["task1"] - g1_t2["task2"] - g1_t3["task3"] - - g1_t1 --> g1_t2 - g1_t1 --> g1_t3 + subgraph extraction["data_extraction"] + e1["extract_source_1"] --> e2["extract_source_2"] --> e3["extract_source_3"] end - - %% group2 - subgraph G2["group2"] - g2_t1["task1"] - g2_t2["task2"] - - g2_t1 --> g2_t2 + subgraph transformation["data_transformation"] + t1["transform_step_1"] --> t2["transform_step_2"] end - - %% group3 - subgraph G3["group3"] - g3_t1["task1"] - g3_t2["task2"] - - g3_t1 --> g3_t2 + subgraph loading["data_loading"] + l1["load_to_target_1"] --> l2["load_to_target_2"] end - - %% зависимости между группами - g1_t2 --> g3_t1 - g1_t3 --> g3_t1 - g2_t2 --> g3_t1 + e3 --> l1 + t2 --> l1 ``` -Визуально в интерфейсе Airflow группы задач отображаются как один узел с небольшим индикатором. Клик по нему разворачивает или сворачивает вложенные задачи, что значительно улучшает восприятие DAG с большим количеством задач и связей, особенно когда в них десятки и сотни задач. - -TaskGroup — это удобный способ логической группировки задач, который помогает упростить код и представить сложные пайплайны более компактно. +Для запуска сохраните пример в `airflow-docker/dags/task_group_example_dag.py`. После ручного запуска все семь задач должны завершиться успешно. Нажмите название группы со стрелкой, чтобы раскрыть ее. Полный идентификатор первой задачи - `data_extraction.extract_source_1`: имя группы становится префиксом `task_id`. Такой идентификатор нужен при обращении к задаче через XCom. ## Система оповещений (алертинг) @@ -285,174 +275,12 @@ def notify_success(context): Проверка: сначала воспроизведите окончательный отказ, затем успешный запуск. Сверьте текст каждого письма с состоянием соответствующего DAG Run. Упражнения находятся в [блоке ошибок](educational-tasks.md#errors) и [задании на почту](educational-tasks.md#resources). -## Практическое применение: улучшенный DAG +## Практика на учебном стенде -Теперь применим изученные концепции для усовершенствования нашего DAG. Добавим переменные и разобьем задачи на логические группы. +Откройте [resource_management_dag.py](airflow-docker/dags/resource_management_dag.py). Он имитирует чтение двух источников и передает небольшие результаты через XCom. Для запуска нужен пул `training_pool` с двумя слотами; создайте его в Admin > Pools. -Сначала создадим переменную `DATABASE_URL` со строкой подключения к базе данных через веб-интерфейс Airflow и импортируем ее в коде DAG: +В исходном DAG задачи `read_customers` и `read_orders` не зависят друг от друга, но занимают два и один слот соответственно. Поэтому они выполняются последовательно. `calculate_metrics` получает их результаты через `xcom_pull`, а `log_metrics` пишет в лог словарь `customers=20`, `orders=30`. -В этом примере используется переменная 'database_connection_string', предварительно созданная в интерфейсе Airflow, для хранения строки подключения к базе данных. Это позволяет избежать жесткого кодирования конфиденциальной информации в коде DAG и упрощает настройку подключения для разных окружений. +В [заданиях 8.1-8.3](educational-tasks.md#resources) вы измените вес задачи в пуле, добавите метрику и объедините задачи чтения в TaskGroup. Там описаны изменения, ожидаемые результаты и места проверки. Файлы CSV и подключение к БД для этого примера не нужны: чтение источников имитируется. -```python -import os -import datetime as dt -import pandas as pd -from airflow.models import DAG -from airflow.operators.bash import BashOperator -from airflow.operators.python import PythonOperator -from airflow.operators.dummy import DummyOperator -from airflow.utils.task_group import TaskGroup -from airflow.models import Variable -from sqlalchemy import create_engine - -DATABASE_URL = Variable.get('database_connection_string') - -args = { - 'owner': 'analytics_team', - 'start_date': dt.datetime(2021, 6, 15), - 'retries': 2, - 'retry_delay': dt.timedelta(minutes=2), -} - -# функции для обработки данных -def get_file_path(file_name): - return os.path.join(os.path.expanduser('~/data'), file_name) - -def load_customer_data(): - file_path = get_file_path('customer_data.csv') - df = pd.read_csv(file_path) - engine = create_engine(DATABASE_URL) - df.to_sql('customers', engine, index=False, if_exists='replace', schema='staging') - -def aggregate_customer_data(): - engine = create_engine(DATABASE_URL) - customer_df = pd.read_sql('select * from staging.customers', con=engine) - - df = customer_df.groupby(['region', 'category']).agg( - total_orders=('orders', 'sum'), - avg_amount=('amount', 'mean') - ).reset_index() - - df.to_sql('customer_summary', engine, index=False, if_exists='replace', schema='analytics') - -dag = DAG( - dag_id='customer_pipeline_enhanced', - schedule_interval=None, - default_args=args, -) -``` - -Теперь разобьем задачи на логические группы и добавим Jinja-шаблоны для доступа к контексту выполнения: - -В этом примере создается начальная задача 'pipeline_start', которая использует Jinja-шаблоны для вывода информации о запуске DAG, включая идентификатор запуска (run_id) и информацию о DAG Run. Затем задачи группируются в логическую группу 'data_processing_stage', что улучшает структуру и читаемость DAG. - -```python -# Начальная задача с информацией о запуске -start_task = BashOperator( - task_id='pipeline_start', - bash_command='echo "Pipeline started! Run ID: {{ run_id }} | DAG Run: {{ dag_run }}"', - dag=dag, -) - -# Группа задач по предварительной обработке данных -with TaskGroup(group_id="data_processing_stage") as data_processing: - # Загрузка данных - load_customer_dataset = PythonOperator( - task_id='load_customer_data', - python_callable=load_customer_data, - dag=dag, - ) - # Агрегация и запись данных - aggregate_customer_dataset = PythonOperator( - task_id='aggregate_customer_data', - python_callable=aggregate_customer_data, - dag=dag, - ) - load_customer_dataset >> aggregate_customer_dataset - -# Установка последовательности выполнения -start_task >> data_processing -``` - -Поскольку последовательность задач внутри групп указывается при их создании, в конце необходимо определить порядок выполнения самих групп, чтобы планировщик понимал общую логику выполнения. - -Итоговый DAG будет выглядеть следующим образом: - -```python -import os -import datetime as dt -import pandas as pd -from airflow.models import DAG -from airflow.operators.bash import BashOperator -from airflow.operators.python import PythonOperator -from airflow.operators.dummy import DummyOperator -from airflow.utils.task_group import TaskGroup -from airflow.models import Variable -from sqlalchemy import create_engine - -DATABASE_URL = Variable.get('database_connection_string') - -args = { - 'owner': 'analytics_team', - 'start_date': dt.datetime(2021, 6, 15), - 'retries': 2, - 'retry_delay': dt.timedelta(minutes=2), -} - -def get_file_path(file_name): - return os.path.join(os.path.expanduser('~/data'), file_name) - -def load_customer_data(): - file_path = get_file_path('customer_data.csv') - df = pd.read_csv(file_path) - engine = create_engine(DATABASE_URL) - df.to_sql('customers', engine, index=False, if_exists='replace', schema='staging') - -def aggregate_customer_data(): - engine = create_engine(DATABASE_URL) - customer_df = pd.read_sql('select * from staging.customers', con=engine) - - df = customer_df.groupby(['region', 'category']).agg( - total_orders=('orders', 'sum'), - avg_amount=('amount', 'mean') - ).reset_index() - - df.to_sql('customer_summary', engine, index=False, if_exists='replace', schema='analytics') - -dag = DAG( - dag_id='customer_pipeline_enhanced', - schedule_interval=None, - default_args=args, -) - -# Начальная задача -start_task = BashOperator( - task_id='pipeline_start', - bash_command='echo "Pipeline started! Run ID: {{ run_id }} | DAG Run: {{ dag_run }}"', - dag=dag, -) - -# Группа предварительной обработки -with TaskGroup(group_id="data_processing_stage") as data_processing: - load_customer_dataset = PythonOperator( - task_id='load_customer_data', - python_callable=load_customer_data, - dag=dag, - ) - aggregate_customer_dataset = PythonOperator( - task_id='aggregate_customer_data', - python_callable=aggregate_customer_data, - dag=dag, - ) - load_customer_dataset >> aggregate_customer_dataset - -start_task >> data_processing -``` - -В этом материале мы рассмотрели расширенные возможности Airflow, которые помогут улучшить работу ваших пайплайнов: -- Управление ресурсами с помощью пулов задач -- Обмен данными между задачами через XCom -- Логическая группировка задач с TaskGroup -- Настройка системы оповещений - -Помните: не стоит использовать все доступные функции сразу. Выбирайте инструменты последовательно и находите оптимальный набор возможностей под конкретную задачу. +Для опытов с callback используйте [error_handling_dag.py](airflow-docker/dags/error_handling_dag.py) и [блок 7](educational-tasks.md#errors). Отправка настоящей почты остается заданием по желанию. diff --git a/docs/specs/2026-09-28-practice-improvement-plan.md b/docs/specs/2026-09-28-practice-improvement-plan.md index aa860f4..9127f9a 100644 --- a/docs/specs/2026-09-28-practice-improvement-plan.md +++ b/docs/specs/2026-09-28-practice-improvement-plan.md @@ -28,12 +28,14 @@ ## Текущее состояние реализации -Исправления DAG сохранены в `cf1b770`, задания и навигация в `d8dfe83`. Последующие уточнения и этот протокол входят в завершающий коммит ветки `work/airflow-practice-refresh`. +Исправления DAG сохранены в `cf1b770`, задания и навигация в `d8dfe83`, уточнения после прохождения в `1277bff`. В `f6b450c` задания перенесены в корень репозитория. Все 25 исходных заданий сопоставлены с 34 упражнениями: 16 основных и 18 по желанию. Основной маршрут пройден без дополнительных упражнений. Затем проверены дополнительные расширения и возвраты к базовой схеме, режимам и настройкам. Сохранены [условия и результаты проверки](../validation/2026-09-28-practice.md), [состояния 70 контрольных и диагностических запусков](../validation/2026-09-28-practice-runs.jsonl) и снимки UI. Открытых этапов реализации нет. При следующей правке сверять затронутые задания с кодом и протоколом. Проверка не измеряет трудность на студентах; SMTP проверен локальным приемником, без внешнего почтового провайдера. +После независимого педагогического ревью исправлены примеры в главах 2, 7 и 9 и уточнены ссылки на теорию перед практикой. Значимость находок, принятые решения и исполнение примеров записаны в [перепроверке ревью](../validation/2026-09-28-theory-review.md). Предложение расширить основную практику опытом с Clear оставлено до проверки на студентах; состав упражнений сохранен. + ## Согласованные решения 1. **Последовательность внутри блока.** Студент запускает исходный пример, меняет одну существенную настройку или часть графа и наблюдает результат. Изменения основных заданий внутри блока сохраняются. diff --git a/docs/validation/2026-09-28-theory-review.md b/docs/validation/2026-09-28-theory-review.md new file mode 100644 index 0000000..5ece226 --- /dev/null +++ b/docs/validation/2026-09-28-theory-review.md @@ -0,0 +1,50 @@ +# Перепроверка педагогического ревью + +Дата: 2026-09-28. Исходная версия: `f6b450c`. После независимого ревью замечания проверены по текстам, Compose и исполнением примеров в Apache Airflow 2.9.2. + +## Оценка значимости и решения + +Основные задания используют готовые DAG и правильные параметры учебной БД. Подтвержденные ошибки относятся к примерам теоретических глав: они мешают студенту, который переносит код из главы в стенд. Оценка их как блокировки всей основной практики была бы завышенной. + +| Находка | Что подтвердилось | Решение | +|---|---|---| +| Параметры Connections из другого стенда | В главе 7 предлагались `postgres / airflow / airflow`, хотя Compose и практика используют `postgres-training / training / student` | Исправлены поля формы. Добавлен полный DAG проверки подключения и наблюдаемый результат в БД | +| Задачи без привязки к DAG | Примеры `greeting_dag` и `template_tracking_example` падают при объявлении зависимостей; `template_example` создает DAG без задач | Добавлены привязки и объяснение. Примеры шаблонов дополнены импортами и настройками, чтобы их можно было запускать целиком | +| Неподготовленный итоговый пример главы 9 | Нет исходного CSV и подготовки схем БД, расходятся имена Variable. После подстановки Variable пример падает при создании TaskGroup | Вместо отдельного пайплайна глава направляет к существующему `resource_management_dag.py` и блоку 8. Пример самой TaskGroup сделан самостоятельным; схема согласована с кодом | +| Слишком широкое чтение перед упражнениями | Для XCom перед блоком 4 была указана вся глава 9 | В таблице маршрута даны ссылки на нужные разделы. Callback отмечен как подготовка к дополнительному 7.B | +| Дополнительный опыт с Clear | Нехватка понимания повторов у студентов не подтверждена. Основные задания уже показывают новые запуски и retry; в 4.1 есть объяснение различия повторной загрузки и нового CSV | Обязательная практика сохранена: 16 основных и 18 дополнительных заданий. Опыт с Clear имеет смысл добавлять после наблюдения за студентами, если они путают эти случаи | + +## Воспроизведение исходных ошибок + +Код извлечен из Python-блоков глав на `f6b450c` и исполнен в одноразовом контейнере образа `airflow-manual-practice-check:2.9.2`, ранее собранного из Dockerfile стенда. Версия внутри контейнера: `2.9.2`. + +В неполные примеры главы 7 переданы недостающие импорты и `default_args`, чтобы проверить именно привязку задач. Для итогового примера главы 9 `Variable.get` заменен на тестовое значение строки подключения: это позволило дойти до создания TaskGroup без подготовки посторонних данных. + +| Пример | Результат до правок | +|---|---| +| `greeting_dag` | `AirflowException: Tried to create relationships between tasks that don't have DAGs yet` | +| `template_example` | Разбор без исключения, список задач пуст | +| `template_tracking_example` | То же исключение при создании зависимостей | +| `customer_pipeline_enhanced` | `AirflowException: TaskGroup can only be used inside a dag` | + +## Проверка исправленных примеров + +Полные Python-блоки текущих глав исполнены без дополнений к коду. Затем вызван `DAG.test()` для логической даты `2024-01-02 UTC`; SQL-пример дополнительно запущен для `2024-01-03 UTC`. + +Окружение: тот же образ Airflow 2.9.2, отдельная SQLite для метаданных и отдельная PostgreSQL 16 с базой `training`. Подключение `my_postgres_conn` передано через окружение с полями из инструкции. Контейнеры работали во внутренней сети без опубликованных портов. Каталог репозитория смонтирован только для чтения. Временные контейнеры, сеть и данные удалены после проверки. + +| Пример | Проверенный результат | +|---|---| +| `greeting_dag` | Две успешные задачи; функция приветствия исполнилась | +| `template_example` | Одна успешная задача; результат Bash - `2024-01-02` | +| `template_tracking_example` | Две успешные задачи; результаты Bash - `2024-01-02` и `Processing for 2024-01-02` | +| `connection_example` | Оба запуска успешны; `to_regclass` возвращает `connection_demo_users`, таблица содержит 0 строк | +| `task_group_example` | Все семь задач успешны; проверены полные идентификаторы и шесть зависимостей, показанных на схеме | + +Все шесть DAG Run завершились успешно. Ссылки на файлы и разделы проверены после преобразования Markdown в HTML. Содержание упражнений после таблицы маршрута совпадает с исходной версией. + +## Границы проверки + +`DAG.test()` подтверждает разбор графа, выполнение задач, подстановку шаблонов и работу SQL. Этот прогон не проверяет работу планировщика по расписанию, форму Connections и отображение графов в UI. Основные DAG стенда не менялись; прежний [протокол практики](2026-09-28-practice.md) остается отдельной проверкой их поведения. + +Трудность и время прохождения на студентах не измерялись. Педагогическое предположение о пользе нового опыта с Clear остается гипотезой, а не открытой ошибкой реализации. diff --git a/educational-tasks.md b/educational-tasks.md index b7c9437..d4296dd 100644 --- a/educational-tasks.md +++ b/educational-tasks.md @@ -10,14 +10,23 @@ | Блок | Примеры | Что наблюдаем | Перед практикой | |---|---|---|---| -| [1. Первый DAG](#first-dag) | `hello_world_dag.py` | Задачи, зависимости, операторы, расписание | Главы 4, 5 и 8 | -| [2. SQL и подключения](#sql) | `sql_basic_dag.py` | Connection, SQL, ошибка подключения | Глава 7 | +| [1. Первый DAG](#first-dag) | `hello_world_dag.py` | Задачи, зависимости, операторы, расписание | Главы 4 и 5; для 1.3 - [расписания][schedules] | +| [2. SQL и подключения](#sql) | `sql_basic_dag.py` | Connection, SQL, ошибка подключения | [Подключения][connections] | | [3. Работа с файлами](#files) | `file_operations_dag.py` | Результаты отдельных задач | Главы 5 и 6 | -| [4. Загрузка и DQ](#load-dq) | `csv_to_postgres.py`, `csv_to_postgres_dq.py` | XCom, данные в БД, ошибка проверки | Главы 6, 7 и 9 | +| [4. Загрузка и DQ](#load-dq) | `csv_to_postgres.py`, `csv_to_postgres_dq.py` | XCom, данные в БД, ошибка проверки | Глава 6, [шаблоны][templates] и [XCom][xcom] | | [5. Разбор ETL](#etl) | `data_processing_dag.py` | Параллельные шаги и путь данных | Глава 5 | -| [6. Ветвление](#branching) | `branching_dag.py` | Variables, выбранная ветка, `skipped` | Главы 6 и 7 | -| [7. Ошибки и повторы](#errors) | `error_handling_dag.py` | Попытки, обработчики, итог запуска | Главы 6 и 9 | -| [8. Пулы и группировка](#resources) | `resource_management_dag.py` | Слоты, XCom, TaskGroup | Глава 9 | +| [6. Ветвление](#branching) | `branching_dag.py` | Variables, выбранная ветка, `skipped` | Глава 6 и [Variables][variables] | +| [7. Ошибки и повторы](#errors) | `error_handling_dag.py` | Попытки, обработчики, итог запуска | Глава 6; для 7.B - [callback][alerts] | +| [8. Пулы и группировка](#resources) | `resource_management_dag.py` | Слоты, XCom, TaskGroup | [Пулы][pools], [XCom][xcom] и [TaskGroup][taskgroups] | + +[schedules]: 08%20-%20Управление%20временем.md#настройка-автоматических-запусков +[connections]: 07%20-%20Шаблоны,%20переменные%20и%20подключения%20в%20Airflow.md#централизованные-подключения-connections +[templates]: 07%20-%20Шаблоны,%20переменные%20и%20подключения%20в%20Airflow.md#динамические-шаблоны-airflow-на-основе-jinja +[variables]: 07%20-%20Шаблоны,%20переменные%20и%20подключения%20в%20Airflow.md#безопасные-переменные-variables +[xcom]: 09%20-%20Пулы,%20XCom,%20TaskGroup%20и%20алертинг%20в%20Airflow.md#обмен-данными-между-задачами-xcom-и-контекст-выполнения +[pools]: 09%20-%20Пулы,%20XCom,%20TaskGroup%20и%20алертинг%20в%20Airflow.md#управление-ресурсами-с-помощью-пулов-задач +[taskgroups]: 09%20-%20Пулы,%20XCom,%20TaskGroup%20и%20алертинг%20в%20Airflow.md#группировка-задач-с-помощью-taskgroup +[alerts]: 09%20-%20Пулы,%20XCom,%20TaskGroup%20и%20алертинг%20в%20Airflow.md#система-оповещений-алертинг Главы доступны в [оглавлении учебника](README.md). Инструкция запуска и параметры учебной БД находятся в [README стенда](airflow-docker/README.md), файлы примеров - в [airflow-docker/dags/](airflow-docker/dags/).