Улучшение DQ в DWH через внедрение Data Quality DAG с блокировкой некорректных данных и алертами #313
Description
Улучшение контроля качества данных в DWH 🚦
Текущая ситуация и её проблемы ❗
Данные из stg, ods, dwh и dm загружаются друг за другом без жёстких проверок качества.
Ошибки и «мусор» чаще всего обнаруживаются уже после попадания данных в основные таблицы.
Хотя алерты есть, плохие данные всё равно успевают просочиться, портя аналитику и усложняя поиск причин.
Отсутствие автоматической блокировки некорректных данных приводит к долгим расследованиям и снижает доверие к хранилищу.
Нет быстрого способа локализовать и выделить проблемные данные для анализа.
Как это решаем? Предлагаемая архитектура 🔧
Создаём отдельный Data Quality DAG, который запускается сразу после загрузки в staging-таблицы.
В этом DAG реализуем задачи на проверку качества данных:
Проверяем формат, полноту, уникальность, соответствие бизнес-правилам — с помощью SQL.
Если проверки проходят — данные коммитятся в ods/dwh.(продолжаются текущие реализованные пайплайны)
Если есть ошибки — проблемные записи перемещаются в отдельную таблицу quarantine для разбора.
Используем ветвление DAG, чтобы при ошибках останавливать загрузку в основные таблицы.
Автоматически отправляем детализированные алерты в Slack/email/Telegram с описанием причин.
Преимущества такого подхода ⭐
🚫 Плохие данные не попадают в DWH — аналитика становится точнее.
⏱️ Ошибки видны и разбираются сразу — легче быстро реагировать.
📦 Есть отдельное хранилище для проблемных данных — удобно анализировать и исправлять.
🔄 Процесс интегрируется в Airflow, легко поддерживается и расширяется.
📊 В дальнейшем можно собрать метрики качества для дашборда и контроля.
Архитектура пайплайна подробно 🛠️
Запуск DAG
После загрузки сырых данных в stg из бекенда, Telegram и GitHub.
Этапы внутри Data Quality DAG
Проверка данных (Data Quality Checks) с помощью SQL:
Проверка на NULL в обязательных полях
Проверка правильности форматов (даты, email и др.)
Проверка уникальности ключей
Проверка бизнес-правил (напр. значения в пределах)
Каждый чек выдаёт Pass/Fail
Агрегация результатов с BranchPythonOperator:
Если все проверки успешны, идём дальше
Если хотя бы одна провалена — запускается обработка ошибок
Если прошло — коммит в ods:
Перенос данных из stg → ods (Insert/Update)
Если ошибки — quarantine:
Копируем проблемные записи в отдельную quarantine таблицу
Храним причины ошибок, время, источник
Можно удалить проблемные данные из stg, не давая им идти дальше
Алерты:
Уведомляем команду через тг
В сообщениях — ссылки и пояснения по ошибкам
Ветвление (BranchPythonOperator)
Позволяет контролировать поток: коммит или quarantine.
Логирование и мониторинг
Полные логи в Airflow UI
Возможно добавление метрик для Data Quality Dashboard (уже в редаше потом собирать будем)