Всем привет! Меня зовут Андрей, я занимаюсь процессами контроля качества данных в DataOffice Ростелекома. В статье поделюсь опытом создания модуля контроля качества данных, с какими трудностями мы сталкивались и как их преодолевали.

Введение

Если данные не помогают бизнесу принимать решения, то хранить их дорого и бессмысленно.  Мы в Ростелекоме используем данные для автоматизации бизнес-процессов, для классической аналитики, машинного обучения и принятия бизнес-решений на всех уровнях управления, поэтому для нас большое значение имеет качество данных.

К нашему хранилищу данных подключено более 300 информационных систем на различных технологиях (CDC, ODBC, API и т.д.). Ежедневно в хранилище загружаются десятки тысяч таблиц. Общий объем хранилища превышает 4 петабайта данных, и с ним работает более 5000 пользователей. Управлять качеством данных в такой масштабной системе невозможно без автоматизации.

Окружение

Данные в нашем хранилище распределены по двум системам хранения: озеро данных (data lake) Hive над Hadoop и MPP база данных Greenplum. Озеро данных используется для хранения супер-больших таблиц, таких как данные трафика широкополосного доступа к сети Интернет, а Greenplum – для всех остальных.

В хранилище используется пакетная загрузка данных, раз в день, еженедельно или ежемесячно.

За оркестрацию доставки пакетов отвечает Airflow, за оркестрацию загрузки таблиц отвечает Datastage.

Вопросы бизнеса, на которые нам было трудно отвечать, и предпосылки создания собственной системы

Все ли вовремя загружается?

Любимый вопрос бизнес-заказчиков: когда данные должны появится в отчете?  Несколько лет назад, когда в хранилище загружалось всего лишь несколько сотен таблиц раз в месяц, отвечать на такие вопросы было просто. В дальнейшем количество таблиц выросло до десятков тысяч, и на этот вопрос стало гораздо сложнее отвечать. Конечно, регулярность загрузки фиксируется в документации, но просматривать каждый раз документацию в поисках ответа неудобно. Нужна была система хранения данных о плановых сроках загрузки всех таблиц и отчетов.

Почему данных меньше, чем ожидалось?  

Пользователи отчетности часто замечали нехватку данных в отчетности за некоторые дни и создавали заявки в службу поддержки. Бывает так, что в данных могут образоваться пропуски – по самым разным причинам данные за определенный день могут не загрузиться. Такие сбои создавали трудности в работе бизнес-пользователей. Нужно было реагировать на пропуски до того, как пользователи начнут работать с отчетом.

Почему данные некорректные?

В процессе выгрузки данных с источника мы обогащаем их информацией о контрольных суммах и количестве строк, однако при загрузке в целевой слой данные могут потеряться и не соответствовать контрольным файлам. Ручной поиск расхождений занимал много времени, ошибки искажали отчетность. Нужно было отлавливать такие ошибки оперативно и не пропускать неполные/искаженные данные дальше.

Почему данные не совпадают с эталонными?

Детальные данные (detailed data store) в хранилище данных Greenplum и агрегированные в ERP (Enterprise Resource Planning, включая бухучет) в базе данных Oracle не всегда сходились. Это снижало доверие к данным в хранилище. Нужно было уметь сравнивать данные из разных систем и баз данных на разных технологиях и делать сверочные отчеты.

С этим нужно что-то делать

Если пользователи периодически сталкиваются с ошибками в данных, это заставляет их сомневаться в корректности остальных данных. Пользователи вынуждены проводить расследования. Это тормозит их работу и мешает принимать решения на основе данных из хранилища. Как итог, они теряют доверие к отчетности в целом и перестают ею пользоваться.

Поэтому критичные ошибки, связанные с качеством данных, не должны доходить для потребителя. Ошибки нужно отлавливать автоматически и оперативно их устранять. А формирование отчетов нужно блокировать, пока ошибки не будут устранены.

При реализации первых проверок мы столкнулись с дополнительными требованиями:

• Все проверки должны выполняться в общем конвейере загрузки данных (data pipeline), не занимая более 10% от общего времени загрузки. А так как у нас уже был разработан свой модуль управления конвейером загрузки (Управляющий механизм) с использование IBM DataStage для запуска заданий, с логикой работы на Python, с настроечными таблицами  и ETL скриптами,  хранящимися   в БД Oracle, готовые решения нам не подходили.

• Писать множество проверок для каждого из десятков тысяч объектов очень дорого, проверки должны быть универсальными, названия проверяемых атрибутов и таблиц нужно передавать в проверку через стандартизованные параметры.

Реализация

Для решения перечисленных проблем был создан отдельный Модуль контроля качества данных (МККД). Модуль работает в тесной связи с Управляющим механизмом (data pipeline orchestration) и позволяет настраивать проверки для запуска по времени или по событию загрузки таблицы на слой данных хранилища.

Особенности реализации:

• МККД может подключаться для проверок к различным БД: Hive (озеро данных), Greenplum (основное хранилище данных), Oracle (внешние системы). Список БД для проверок может быть расширен с помощью несложной настройки.

• Модуль реализован как черный ящик с открытым API, которое используется как в data pipeline orchestration, так и в других механизмах, а также позволяет без проблем запускать проверки вручную.

• Проверкой (check) называется именованный абстрактный объект, к которому привязаны один или несколько SQL-запросов, правила обработки результата для каждого запроса, параметры, конфигурации подключения к ХД, список объектов ХД, к которым можно применить проверку и т.д. Всё это хранится в настроечных таблицах отдельной БД. Такая стандартизация значительно упрощает разработку и тиражирование проверок.

• Для выполнения проверки вызывается API, которое создает задание на проверку (checktask). Задание охарактеризовано уникальным идентификатором, конкретным объектом ХД и проверкой, которую надо выполнить, а также возможно бизнес-периодом, которым надо ограничить проверяемые данные – не проверять же фактовые таблицы целиком, когда загружался всего один новый день. Новое задание попадает в очередь FIFO, из которой оркестратор выбирает первые несколько штук и отправляет на выполнение. Наличие очереди позволяет балансировать нагрузку на БД хранилища (в первую очередь на Greenplum), т.к. оркестратор гибко вычисляет число заданий на проверку, которое можно в моменте отправить в работу без риска перегрузить БД. За выполнение проверки отвечает универсальный python-скрипт, который принимает на вход параметры Имя_проверки + Слой + Имя_таблицы + Проверяемый_бизнес_период. Скрипт идет в БД за настройками подключения и SQL-запросами, при необходимости подставляет параметры в запрос(-ы), отправляет собранный запрос в нужную БД, обрабатывает результат, формирует финальный статус (success, error, warning) и сохраняет его в универсальном для всех проверок журнале.

Пример простой проверки на дубли

Проверка универсальная, т.е. применима к нескольким объектам, и содержит один SQL-запрос:

select '~TABLE~' as table_name,
       'Проверка составного ключа на уникальность' as description,
	 0 as plan_cnt,           -- ожидаемый показатель
       fact.cnt as fact_cnt,       -- фактический показатель
	 case when 0=fact.cnt then 'Критерии сверки выполнены' else 'Критерии сверки нарушены' end as script_status,
        to_char(current_timestamp, 'YYYY-MM-DD HH24:MI:SS') as load_dttm 
  from (select count(*) cnt 
          from (select ~PK~, count(*) 
		    from  ~LAYER~.~TABLE~ 
		   group by ~PK~
		  having count(*) > 1 ) x
        )  fact;

С параметрами ~TABLE~ - имя таблицы, ~PK~ - перечень первичных ключей, ~LAYER~ - слой хранилища данных, схема Greenplum.

Python-скрипт получает на вход значения ~LAYER~ и ~TABLE~, ~PK~ находит самостоятельно на основе предустановленных правил,  подставляет параметры и спускает SQL в Greenplum. Результат сохраняется в таблице результатов в БД Oracle.

Все проверки разделены по следующим типам ошибок в данных:

• Технические, блокирующие. Например, наличие дублей в данных, которые могут привести к искажению данных в отчетности. Такие ошибки блокируют загрузку конвейера данных. После их оперативного исправления загрузка продолжается.

• Бизнесовые, неблокирующие. Специализированные проверки для определенной предметной области. Например, проверка заполняемости каналов продаж в отчете по продажам. Результаты таких проверок не блокируют загрузку и направляются бизнес пользователям для корректировки данных в системах источниках либо для корректировки бизнес процессов.

За счёт того, что процесс унифицирован – время на создание новых проверок занимает не больше спринта. Новые проверки разрабатываются при создании отчетности. В процессе разработки с заказчиком проговаривают все требования к качеству данных, проверяемые сущности, типы проверок и критерии успешности. Но бывают ситуации, когда приходится добавлять дополнительные проверки к уже действующим отчетам, если обнаруживаются новые ошибки и проблемы с качеством данных. 

Результаты

То, что нельзя измерить, нельзя улучшить, поэтому для постоянного улучшения уровня качества данных были разработаны численные показатели – метрики.

Актуальность. Показывает, насколько фактические даты загрузки данных в хранилище отстают от плановых. Для расчета данной метрики была создана таблица, хранящая плановый сроки загрузки по всем объектам хранилища. Метрика доступна как по каждому объекту, так и суммарно по слоям данных.

После внедрения планового расписания в 2020 году заметно снизился процент пакетов, пришедших с отставанием от расписания. При общем росте числа объектов в 10 раз на 30% снизился показатель опоздавших пакетов, как показано на графике ниже.

Актуальность измеряется в абсолютных и в относительных единицах. Например, среднее отставание от плана для ежедневных витрин – 5 часов. Или исполнение ежедневного регламента за прошлый день по агрегированному слою данных на уровне 57% от норматива.

Актуальность улучшилась за счет оперативного выявления опозданий, анализа причин и устранения часто возникающих проблем.

Уровень качества данных. Показывает количество ошибок за период, в абсолютном выражении.

Настроенные системы проверок качества данных помогают нам оперативно реагировать на проблемы и устранять многие ошибки превентивными мерами. Благодаря этому при постоянном росте количества объектов в хранилище мы сохраняем отрицательную динамику количества ошибок в данных. Достигается это в том числе благодаря настроенному мониторингу, о котором пойдет речь ниже.

Мониторинг

Мониторинг метрик и ошибок ККД осуществляется командой второй линии технической поддержки (2ЛТП). Для мониторинга используется Grafana, письма с оповещениями. Команда 2ЛТП отслеживает типичные ошибки, реагирует и устраняет их.

Какие ошибки отрабатываются в процессе мониторинга?

Технические ошибки, возникающие в процессе ETL: дубли, потери данных между слоями и т.д. Первоначально при включении проверок таких ошибок были сотни, сейчас их единицы. Они быстро выявляются, а причины их возникновения устраняются. Как правило, это дефекты разработки отчетности продуктовыми командами.

Пропуски данных, одного дня или нескольких периодов. Уровень таких ошибок уменьшился с несколько сотен до ста в месяц, но главное — благодаря ККД выросла оперативность устранения таких ошибок.

Трендовые и другие проверки бизнес показателей в отчетности. Такие ошибки становятся основными, они вызваны некорректными данными в системах источниках. Для устранения таких ошибок делаются автоматические рассылки писем в филиалы. Письма содержат описания ошибки и сроки их устранения.

В мониторинг включен контроль сроков исправления ошибок на стороне источников и оповещение пользователей отчетности о проблемах.  

Планы на будущее

Сейчас у нас настроены процессы мониторинга контроля качества данных и решены проблемы качества на уровне интеграций, ETL и хранилища. Но проблему качества на уровне источников данных пока решить до конца не удалось. Исправлять ошибки намного дешевле до их возникновения. Поэтому мы планируем создавать новые проверки для контроля качества данных на системах источниках. Например, контроль заполнения каналов продаж операторами. Результаты этих проверок адресно будут рассылаться на ответственных в филиалы.

Также мы взяли курс на использование инструментов профилирования. Существующие решения по профилированию данных интегрируются с системами Управления данных, а т.к. у нас уже разработана своя система управления данными, то инструмент профилирования мы тоже решили разработать свой. Инструменты профилирования позволят выявлять аномалии в данных до их загрузки в хранилище, а также быстрее проводить анализ данных при подключении новых систем.