docs(practice): уточнить инструкции по результатам прохождения
Уточнены порядок CSV/COPY, расположение логов callbacks, шаблонов и управления TaskGroup. Сохранены условия проверки, результаты 69 контрольных запусков и одного диагностического, SHA-256 DAG и снимки UI. Все этапы плана закрыты. Основной и дополнительный маршруты проверены на Airflow 2.9.2. SMTP проверен локальным приемником, включая отказ доставки; внешняя почта и Windows отдельно не проверялись. Тестовые контейнеры и тома удалены.
This commit is contained in:
ddadmin committed
1 parent
d8dfe83651
commit
1277bff690
10 files changed
+240
-51
No files matched your search
@@ -16,7 +16,9 @@ def choose_format():
|
||||
file_format = Variable.get("branch_format", default_var="csv")
|
||||
branches = {"csv": "process_csv_branch", "json": "process_json_branch"}
|
||||
if file_format not in branches:
|
||||
raise ValueError(f"branch_format должен быть csv или json, получено: {file_format}")
|
||||
raise ValueError(
|
||||
f"branch_format должен быть csv или json, получено: {file_format}"
|
||||
)
|
||||
logging.info("Выбран формат: %s", file_format)
|
||||
return branches[file_format]
|
||||
|
||||
|
||||
@@ -121,6 +121,8 @@ def _load_csv(csv_path: str) -> None:
|
||||
cur.execute(
|
||||
"CREATE TEMP TABLE tmp_orders (LIKE public.orders INCLUDING DEFAULTS) ON COMMIT DROP;"
|
||||
)
|
||||
# Порядок полей CSV должен совпадать со списком COPY.
|
||||
# HEADER пропускает заголовок, а не сопоставляет имена колонок.
|
||||
cur.copy_expert(
|
||||
"COPY tmp_orders (order_id, order_ts, customer_id, amount) FROM STDIN WITH CSV HEADER",
|
||||
f,
|
||||
|
||||
@@ -165,7 +165,7 @@ ALTER TABLE public.students_sample ADD COLUMN email VARCHAR(100);
|
||||
|
||||
После `transform_data` поставьте три параллельные задачи. Каждая читает полный `processed_data.csv`, отбирает свой отдел, проверяет уникальность `id` и сохраняет отдельный файл. После них добавьте сборку этих трех файлов и проверьте количество строк и уникальность `id` уже в общем результате. Не перезаписывайте исходный файл из параллельных задач.
|
||||
|
||||
Для опыта с повтором одна из задач должна намеренно падать только на первой попытке; используйте `ti.try_number` и `retries=1`. Этот прием разобран в блоке 7.
|
||||
Для опыта с повтором одна из задач должна намеренно падать только на первой попытке; используйте `ti.try_number`, `retries=1` и `retry_delay=timedelta(seconds=2)`. Этот прием разобран в блоке 7.
|
||||
|
||||
**Что должно получиться:** объединенный результат содержит те же 100 уникальных `id`. В графе три задачи имеют общего предшественника, но не зависят друг от друга. Повтор не добавляет дубликаты: файл отдела перезаписывается.
|
||||
|
||||
@@ -179,7 +179,7 @@ ALTER TABLE public.students_sample ADD COLUMN email VARCHAR(100);
|
||||
|
||||
Основная практика. Сложность: начальная.
|
||||
|
||||
Запустите `csv_to_postgres`. В задаче `generate_csv` откройте вкладку XCom с ключом `return_value`: там путь к CSV. Найдите этот файл в `data/output`. Затем выберите `load_csv_to_postgres`, откройте Details и раздел Rendered Templates. В `opKwargs` должен быть тот же путь, подставленный вместо шаблона параметра `csv_path`.
|
||||
Запустите `csv_to_postgres`. В задаче `generate_csv` откройте вкладку XCom с ключом `return_value`: там путь к CSV. Найдите этот файл в `data/output`. Затем выберите `load_csv_to_postgres`, откройте Details и прокрутите до Rendered Templates. В `opKwargs` вместо шаблона параметра `csv_path` появился путь к файлу. Часть пути в UI может выглядеть как `***`; сравните имя CSV с XCom.
|
||||
|
||||
Проверьте результат в БД:
|
||||
|
||||
@@ -215,7 +215,7 @@ WHERE order_id = (SELECT MIN(order_id) FROM public.orders);
|
||||
|
||||
По желанию. Сложность: высокая. Требуется завершенный основной блок 4.
|
||||
|
||||
Добавьте `status` со значениями `NEW`, `PROCESSING`, `COMPLETED`. Поле нужно провести через четыре участка: генератор CSV, DDL целевой таблицы, список колонок COPY и обе части итогового `INSERT ... SELECT` в `_load_csv()`.
|
||||
Добавьте `status` со значениями `NEW`, `PROCESSING`, `COMPLETED`. Поле нужно провести через четыре участка: генератор CSV, DDL целевой таблицы, список колонок COPY и обе части итогового `INSERT ... SELECT` в `_load_csv()`. Во всех списках добавляйте его в конец. Порядок колонок CSV должен совпадать с COPY: `CSV HEADER` пропускает первую строку, но не сопоставляет поля по ее именам.
|
||||
|
||||
В DDL используйте `status VARCHAR(20) NOT NULL DEFAULT 'NEW'`. Существующую таблицу подготовьте один раз:
|
||||
|
||||
@@ -228,7 +228,7 @@ ADD COLUMN status VARCHAR(20) NOT NULL DEFAULT 'NEW';
|
||||
|
||||
**Что должно получиться:** новые заказы содержат статусы из CSV, старые получили `NEW`, проверка схемы проходит. Проверяйте значения, а не только успешность задачи: забытый `status` в INSERT может оставить всем строкам значение по умолчанию.
|
||||
|
||||
Завершив опыт, верните исходные четыре списка/описания колонок в загрузчике и список схемы в DQ. Затем выполните `ALTER TABLE public.orders DROP COLUMN status;` и запустите DQ. Проверку положительной суммы из 4.2 сохраните.
|
||||
Завершив опыт, удалите `status` из генератора, DDL, COPY и обеих частей INSERT ... SELECT. Уберите его и из `EXPECTED_ORDERS_SCHEMA` в DQ. Затем выполните `ALTER TABLE public.orders DROP COLUMN status;` и запустите DQ. Проверку положительной суммы из 4.2 сохраните.
|
||||
|
||||
### 4.B. Заменить PythonOperator на PostgresOperator
|
||||
|
||||
@@ -286,11 +286,11 @@ ADD COLUMN status VARCHAR(20) NOT NULL DEFAULT 'NEW';
|
||||
|
||||
По желанию. Сложность: высокая.
|
||||
|
||||
Между преобразованием и `simulate_load` добавьте проверку: результат непустой, `customer_id` и `name` заполнены, суммы положительны. В `create_sample_data()` временно укажите для одного заказа несуществующий `customer_id`. Настройте `on_failure_callback`, который пишет в лог идентификатор задачи и причину ошибки; образец есть в задании 7.B.
|
||||
Между преобразованием и `simulate_load` добавьте проверку: результат непустой, `customer_id` и `name` заполнены, суммы положительны. Для этой задачи задайте `retry_delay=timedelta(seconds=2)`, сохранив один повтор. В `create_sample_data()` временно укажите для одного заказа несуществующий `customer_id`. Настройте `on_failure_callback`, который пишет в лог идентификатор задачи и причину ошибки; образец есть в задании 7.B.
|
||||
|
||||
После `generate_report` добавьте копирование `transformed_data.csv` в архивный файл с датой и временем в имени. Для этого подойдет `shutil.copy`.
|
||||
|
||||
**Что должно получиться:** после исчерпания повторов проверка падает, последующие шаги получают `upstream_failed`, новый архив не появляется. После возврата клиента к исходному значению весь DAG успешен, а содержимое нового архива совпадает с результатом запуска. Логи callback ищите в логах scheduler, а не только в логе задачи.
|
||||
**Что должно получиться:** после исчерпания повторов проверка падает, последующие шаги получают `upstream_failed`, новый архив не появляется. После возврата клиента к исходному значению весь DAG успешен, а содержимое нового архива совпадает с результатом запуска. Сообщение callback этой проверки находится в логе ее последней попытки.
|
||||
|
||||
<a id="branching"></a>
|
||||
|
||||
@@ -370,11 +370,7 @@ ADD COLUMN status VARCHAR(20) NOT NULL DEFAULT 'NEW';
|
||||
|
||||
Добавьте функцию `notify_failure(context)` и укажите ее в `on_failure_callback` рабочих задач. В ней используйте `logging.error` и выведите `context['ti'].task_id` и `context['exception']`. В разных опытах бросьте `ConnectionError`, `ValueError` и `OSError` с различимыми сообщениями. Настройка настоящей почты для этого задания не нужна.
|
||||
|
||||
**Что должно получиться:** после исчерпания повторов callback сообщает, какая задача и почему упала. При временной ошибке с последующим успехом он не вызывается. Сообщения callback проверяйте в логах scheduler:
|
||||
|
||||
```bash
|
||||
docker compose logs airflow-scheduler
|
||||
```
|
||||
**Что должно получиться:** после исчерпания повторов callback сообщает, какая задача и почему упала. При временной ошибке с последующим успехом он не вызывается. В этом примере callback назначен оператору: откройте Logs упавшей задачи и выберите последнюю попытку. После сообщения об отказе найдите свой текст из callback.
|
||||
|
||||
Если нужно наблюдать каждый повтор, для этого есть отдельный `on_retry_callback`. Верните исходные исключения и `error_mode=retry_once` перед переходом дальше.
|
||||
|
||||
@@ -410,7 +406,7 @@ docker compose logs airflow-scheduler
|
||||
|
||||
После группировки полные идентификаторы станут `extract.read_customers` и `extract.read_orders`. Обновите обе ссылки в `calculate_metrics()`.
|
||||
|
||||
**Что должно получиться:** в графе появилась сворачиваемая группа, внутри видны две задачи, а метрика по-прежнему равна 50. TaskGroup организует граф; ограничения пула продолжают действовать.
|
||||
**Что должно получиться:** в графе появилась сворачиваемая группа. Нажмите на ее название со стрелкой, чтобы увидеть две задачи внутри. Метрика по-прежнему равна 50; ограничения пула продолжают действовать.
|
||||
|
||||
### 8.A. Подставить XCom в шаблон
|
||||
|
||||
|
||||
Reference in new issue
Block a user