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

Хранилище данных — сердце Транспортного комплекса

С 2013 года Москва стремительно росла, улучшалась и расширяла свои границы. Логично, что  это влекло за собой бурное развитие Транспортного комплекса, а вместе с ним увеличивалось и количество обслуживающих его IT-систем и данных, которые эти системы генерируют.

Для реализации концепции data driven нужно было собрать все эти данные в одном месте, а для этого была нужна отдельная команда и отдельное подразделение. С этого в 2017 году и началась история ИЦ, сердцем которого является хранилище данных. 

На данный момент данные хранилища Инновационного Центра используются во множестве информационно-аналитических продуктов.  В их числе — интерактивная аналитическая отчётность, цифровое мастер планирование города, Экологическая карта, Коммуникационная платформа, Карта аварийности, Справка загруженности и другие решения, направленные на улучшение и развитие Транспортного комплекса Москвы.

К ХД сейчас подключено 45+ систем, которые ежедневно поставляют более 50 Гб обезличенных данных транспортных систем в более чем 700 таблиц. Среди этих данных:

• валидации билетов в городском транспорте, в т.ч. на речных маршрутах

• фотовидеофиксации транспортных средств на дорогах города 

• телематические данные транспортных средств: наземного городского, грузового транспорта, каршеринга, такси, техники ЖКХ, кикшеринга

•  данные транспортных приложений и многое другое. 

Часть из этих данных поступает в ХД в режиме реального времени, например, телематика и фотовидеофиксации. Хранилище реализовано с помощью следующих инструментов: HDFS, Hive, Airflow, Spark.

На самом деле, объём данных рос почти в геометрической прогрессии и, если на заре создания центра  всё развивалось и поддерживалось руками 20 человек, то сейчас над созданием всех этих продуктов работает уже более 150 человек.

Как была выстроена работа с Data quality с начала существования Инновационного Центра

2017 год: необходимость централизовать процессы работы с данными

В начале пути источников было совсем мало и вопрос о системе мониторинга качества данных тогда вообще не стоял, поэтому качество данных  проверяли аналитики DWH периодически вручную в Hive, запуская sql-скрипты, перечень которых вели в Google Sheets.

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

Развивать дальше Хранилище данных без выстроенных процессов DQ было бы безрассудно, поэтому было решено:

1. Автоматизировать запуск проверок с помощью движка на Spark, который каждый день берёт из Google Sheet  все sql-скрипты, написанные вручную аналитиками DWH, и запускает их в ночь. 

2. Результаты запуска складывать в Vertica и визуализировать в Tableau.

3. Оповещения о проблемах направлять на корпоративную почту.

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

• на часть таблиц скрипты вообще не были написаны 

• часть скриптов содержали логические ошибки 

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

• выполнение части скриптов накладывалось на обновление данных 

• мы получали результаты качества данных с большими задержками, иногда достигающими 2-е суток 

• результаты отображались в Tableau и требовали постоянной доработки бордов

• оповещения приходили на почту, которую нельзя было подключить к телефону 

• на ведение рутинного процесса мониторинга качества данных пришлось найти отдельных сотрудников.

2020 год: увеличение источников данных и продуктов в Инновационном центре и необычные кейсы

2020 год стал переломным для Инновационного центра. Вместе с пандемией пришло множество ad-hoc задач по подготовке различной транспортной отчётности, на базе которой постепенно родились такие проекты, как Платформа коммуникаций с пользователями Транспортного комплекса, realtime аналитическая система - Справка загруженности, аналитические системы - Карта ДТП, Экологическая карта, Фабрика данных, автоматизация ежедневной отчётности и множество других. Со всеми этими проектами стало поступать всё больше данных в ХД.

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

• Небольшой процент поездок авто происходили где-то далеко за её пределами, например, в Мексике или в другом месте из-за периодических сбоев работы GPS.

• Однажды самокаты как будто катались по Москве, когда на улице снегопад и 20-ти градусный мороз, хотя на самом деле они были убраны с улиц. Такая ситуация возникла из-за того, что самокаты перевозили с одного места хранения в другое, а они передавали телематику.

• Средняя длительность парковочной сессии в одном из районов Москвы могла быть 25 суток из-за того, что пользователь уехал в отпуск на длительный срок.

Таких примеров можно привести очень много.

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

2022: Первые шаги перехода к автоматическим проверкам

В 2022 году нам пришлось попрощаться с Tableau. К тому времени наш процесс расчёта проверок качества данных стал работать по 8-12 часов в день из-за обилия расчётов, мешая остальным продуктивным процессам и выдавая на выходе супер некачественные данные, которые сотрудники не успевали разобрать даже за 1 день. А на следующий день уже возникали новые некачественные результаты. В общем, ситуация была плачевная.

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

Начали мы с того, что попробовали понять, какие проверки мы сможем создавать автоматически, опираясь на метаданные hive metastore (названия таблиц/атрибутов/партиций).

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

4 вида базовых проверок:

• наличие записей в таблице

• незаполненные обязательные поля

• дубли

• выбросы

  

Для обозначения вхождения атрибута в ключ мы выбрали обозначение PK, для обозначения обязательности заполнения атрибута мы выбрали обозначение NotNull (изображение 5). Для удобного парсинга этого комментария используем разделитель  между признаками и описанием атрибута.

Далее при парсинге DDL из Hive происходит поиск нужных атрибутов по их комментариям, и они подставляются в соответствующее место скрипта проверки DQ.

Таким образом, мы решили сразу 3 самых главных проблемы:

• исключили возможность появления таблиц без основных проверок,

• исключили возможность появления логических ошибок в основных проверках,

• значительно уменьшили объем monkey job.

  Решение вопроса задержки получения информации

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

Пришла идея в конец каждого DAG загрузки или обновления таблицы встроить стандартную Task (задачу), которая будет запускать все проверки, относящиеся к этой таблице за ее период или дату загрузки или обновления. Task запускается моментально после загрузки данных и не блокирует выполнение следующих DAG’ов.

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

Сейчас процесс разработки и передачи задач  между инженерами данных и аналитиками данных устроен так, что невозможно вывести в прод DAG без настроенной DQ Task.

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

Решение вопроса автоматического обновления комментариев DDL

Далее пришло время решить проблему того, что при изменении структур таблиц в части появления новых или удаления старых атрибутов, изменения ключевых или обязательных атрибутов, нам придётся постоянно руками инженеров данных обновлять комментарии в DDL, что выглядело, как новая monkey job.

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

Делаем парсинг этого документа и формируем структурированный комментарий. Если реальный комментарий в Hive отличается от вновь сформированного, то обновляем его. Заодно сравниваем и типы данных на предмет отличия их в документации и в хранилище.

Таким образом, мы получили полностью автоматически обновляемые основные проверки DQ, которые пересобираются без участия человека при любых изменениях в структуре или свойствах таблиц. Прощай, очередная monkey job, которую сами себе чуть не создали. 

Ну, и в качестве бонуса, упростили жизнь инженерам данных при создании/обновлении ddl таблицы - ведь теперь она автоматом генерируется сразу после написания документации.

Решение задачи информирования о проблемах 

К тому моменту Telegram и его боты активно использовались другими компаниями, поэтому мы решили не изобретать велосипед: 

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

Затем мы начали на уровне того же Google Sheet с атрибутами и таблицами вести ещё одно поле с телеграм-никами ответственных сотрудников, которые подставлялись ботами в уведомления с выявленными проблемами.

Таким образом, каждый специалист состоит только в боте продукта/проекта, которым он занимается и реагирует в этом боте только на те сообщения, в которых тэгается именно он.

Пример технического сообщения от бота (изображение 14)

Пример бизнесового сообщения от бота (изображение 15)

Таким образом, были решены все изначально обозначенные проблемы! Ведь теперь:

Основные самые важные проверки качества данных создаются сами на основе документации, которую и так разрабатывает аналитик перед постановкой задачи дата-инженеру.

Затем эти проверки сами начинают выполняться автоматически, как только дата-инженер выкатывает в прод даг обновления таблицы.

 Оповещения по непройденным проверкам сами приходят всем заинтересованным в телефон сразу же, как возникает проблема.

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

Представим, что у нас давно существует отчёт Статистика по выполнению транспортной работы в наземном городском транспорте и DDL для соответствующей таблицы, в которой было 10 атрибутов, один из которых ключевой (см. шаг 1, изображение 16). 

Нам понадобилось добавить в этот отчёт ещё два атрибута - “Перевозчик” и “Плановое количество транспортных средств”, а один - “Плановое количество водителей”- удалить . Для этого аналитик добавляет новые строки для каждого нового атрибута в документации, задаёт им необходимые свойства - это оранжевые строки. Строки с ненужным атрибутом - удаляет - это красная строка (см. шаг 2, изображение 16). После всех перечисленных действий документация актуализирована. Тогда аналитик ставит задачу в таск-трекере дата-инженеру (см. шаг 3, изображение 17). 

К тому моменту на основе скорректированной документации уже сгенерировалась новая DDL для нового атрибутного состава отчёта (см. шаг 4, изображение 17): в DDL уже имеются все новые атрибуты “Перевозчик” и “Плановое количество транспортных средств”, и для них сгенерирован комментарий со всеми свойствами. При этом видим, что атрибут “Перевозчик” вошёл в составной ключ, и это должно повлечь изменение всех проверок, завязанных на ключевых атрибутах, например, проверок на дубли и выбросы. А вот атрибут “Плановое количество водителей” исчез.

Дата-инженер берёт эту DDL,  накатывает на dev контур и  разрабатывает алгоритмы заполнения новых атрибутов. Как только новая DDL появилась в базе данных dev контур, на её основе тут же перегенерировалось множество скриптов стандартных базовых проверок для этой таблицы (см. шаг 5, изображение 17). 

Мы видим, что новый ключевой атрибут “Перевозчик” теперь участвует в проверке на дубли, также он появился в проверке на обязательные атрибуты. Для него сгенерировались отдельные автоматические проверки на лишние, потерянные ключи и на проверку доменов значений.  На новый атрибут “Плановое количество водителей” тоже сгенерировались проверки на выбросы и на диапазоны значений. А вот для атрибута “Плановое количество водителей” все проверки исчезли, потому что его нет в DDL. При этом в истории результатов проверок остались результаты для старого атрибутного состава.

Как только дата-инженер передаст эту доработку на тестирование аналитику, который её примет, она будет закачана в продакшн. Обновлённый пул проверок  для этого отчёта сразу начнёт выполняться в проде, а неудовлетворительные результаты выполнения этих проверок тут же появятся в боте (см. шаг 6, изображение 17).

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

Выводы

1. Централизация данных. Создание Инновационного центра стало ответом на необходимость централизовать обработку больших объемов данных, поступающих из более чем 45 различных ресурсов. Это сделало управление данными эффективнее и обеспечило возможность построения аналитических продуктов.

2. Автоматизация процессов. Переход от ручной проверки качества данных к автоматическому процессу стал необходимым ввиду роста объема и сложности данных. Это изменение позволило обеспечить стабильность и ускорение в обработке информации.

3. Использование современных технологий. В работе с данными были применены технологии HDFS, Hive, Airflow и Spark, что способствовало более эффективному управлению данными и их валидации.

4. Тестирование и использование самых разных современных инструментов. Внедрение современных инструментов, таких как Telegram для уведомлений и мониторинга состояния систем, позволило оперативно реагировать на возникающие проблемы и проводить тестирование в реальном времени.

5. Снижение трудозатрат. Ручной труд в проверке данных был трудоемким и неэффективным. Автоматизация процессов позволила сократить время и ресурсы, затрачиваемые на проверки, что освободило аналитиков для выполнения более важных задач.

6. Качество данных - приоритет для Инновационного Центра. Постоянное улучшение качества данных является ключевым аспектом работы центра, что непосредственно влияет на принятие информированных управленческих решений.