ddadmin
8028711f0c
- Зачем: - адаптация пайплайна для работы с чистым PostgreSQL вместо Greenplum - обеспечение совместимости с учебным стендом на базе Docker Compose - Что: - переименованы файлы (greenplum -> postgres) - обновлены DAG-файлы: замена get_gp_conn на get_postgres_conn, адаптация DDL под PostgreSQL - обновлён helpers/postgres.py: замена функций и переменных окружения - добавлен pytest в requirements.txt для запуска тестов - создан tests/conftest.py с функцией patch_postgres_hook - обновлён tests/test_postgres_helpers.py: исправлены импорты - добавлено монтирование папки tests в docker-compose.yml - адаптирован sql/base/orders_ddl.sql под PostgreSQL - Проверка: - docker-compose exec airflow-webserver pytest /opt/airflow/tests/test_postgres_helpers.py
97 lines
3.6 KiB
Python
97 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
|
|
from airflow.operators.python import PythonOperator
|
|
from helpers.postgres import (
|
|
assert_orders_have_rows,
|
|
assert_orders_no_duplicates,
|
|
assert_orders_schema,
|
|
assert_orders_table_exists,
|
|
get_postgres_conn,
|
|
)
|
|
|
|
from airflow import DAG
|
|
|
|
|
|
def _run_check(check_callable):
|
|
"""
|
|
Оборачивает проверку качества данных в контекст подключения к Postgres.
|
|
|
|
Этот DAG предназначен для автоматической проверки качества данных
|
|
после CSV-пайплайна в таблице public.orders:
|
|
1. Проверяет существование таблицы
|
|
2. Проверяет соответствие схемы
|
|
3. Проверяет наличие данных
|
|
4. Проверяет отсутствие дубликатов
|
|
|
|
Args:
|
|
check_callable: Функция проверки, принимающая подключение к БД
|
|
"""
|
|
# Получаем имя функции для логов
|
|
check_name = check_callable.__name__.replace("assert_", "")
|
|
logging.info("🚀 Запуск проверки: %s", check_name)
|
|
|
|
with get_postgres_conn() as conn:
|
|
check_callable(conn)
|
|
|
|
logging.info("✅ Проверка пройдена: %s", check_name)
|
|
|
|
|
|
def _log_dq_summary():
|
|
"""
|
|
Логирует итоговую сводку по качеству данных.
|
|
Эта задача выполняется после всех проверок и показывает общий результат.
|
|
"""
|
|
logging.info("🎉 Все проверки качества данных пройдены успешно!")
|
|
logging.info("📊 Качество данных в таблице orders соответствует требованиям.")
|
|
|
|
|
|
default_args = {"owner": "airflow", "retries": 1, "retry_delay": timedelta(seconds=30)}
|
|
|
|
with DAG(
|
|
dag_id="csv_to_postgres_dq",
|
|
start_date=datetime(2017, 1, 1),
|
|
schedule=None,
|
|
catchup=False,
|
|
default_args=default_args,
|
|
tags=["demo", "postgres", "quality", "csv", "dq"],
|
|
description="Проверки качества данных после CSV → public.orders в Postgres",
|
|
) as dag:
|
|
# Задача 1: Проверка существования таблицы
|
|
check_exists = PythonOperator(
|
|
task_id="check_orders_table_exists",
|
|
python_callable=_run_check,
|
|
op_args=[assert_orders_table_exists],
|
|
)
|
|
|
|
# Задача 2: Проверка соответствия схемы таблицы
|
|
check_schema = PythonOperator(
|
|
task_id="check_orders_schema",
|
|
python_callable=_run_check,
|
|
op_args=[assert_orders_schema],
|
|
)
|
|
|
|
# Задача 3: Проверка наличия данных
|
|
check_has_rows = PythonOperator(
|
|
task_id="check_orders_has_rows",
|
|
python_callable=_run_check,
|
|
op_args=[assert_orders_have_rows],
|
|
)
|
|
|
|
# Задача 4: Проверка отсутствия дубликатов
|
|
check_no_duplicates = PythonOperator(
|
|
task_id="check_order_duplicates",
|
|
python_callable=_run_check,
|
|
op_args=[assert_orders_no_duplicates],
|
|
)
|
|
|
|
# Задача 5: Итоговая сводка
|
|
dq_summary = PythonOperator(
|
|
task_id="data_quality_summary",
|
|
python_callable=_log_dq_summary,
|
|
)
|
|
|
|
# Определяем последовательность выполнения задач
|
|
check_exists >> check_schema >> check_has_rows >> check_no_duplicates >> dq_summary |