Меня зовут Мешкова Анна. Я руковожу стримом «Озеро данных» на Фабрике данных ВТБ. И этот текст о том, как мы в банке смогли выстроить унифицированный процесс загрузки данных, какие задачи себе ставили и что из этого получилось.

Наша команда входит в Фабрику данных банка ВТБ и отвечает за загрузку всех внешних и внутренних систем-источников в Datalake на базе Hadoop.

Объёмы создаваемых банком данных неуклонно растут: уже сейчас в озере несколько петабайт данных, объём загрузки за последний год вырос в пять раз. За последний год было загружено более 50 новых источников. И каждый день процесс доступа к данным становится на пути решения бизнес-задач. Приоритеты бизнеса сдвигаются в сторону скорости доставки данных и удобства доступа к ним.

Уже недостаточно иметь гарантированную доставку данных. Сейчас надо обеспечивать доставку real time, сохраняя при этом структурированность, надежность и консистентность данных.

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

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

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

Какую цель мы себе поставили?

• Создать единый централизованный механизм загрузки из реляционных источников данных (Oracle DB, MSSQL, PostgreSQL, Vertica, Teradata, HDFS, Arenadata DB) и из нереляционных (S3, Kafka, NFS);

• Новый механизм загрузки должен быть создан на импортозамещенных и opensource решениях;

• Уменьшить количество используемых и поддерживаемых алгоритмов загрузки;

• Упростить процесс разработки (привести к режиму конфигурации);

• Упростить процесс локализации ошибок;

• Перейти к централизованной установке версий.

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

После детального анализа было решено сделать реализацию в виде Spark-приложения на Scala: этот язык — «родной» для Spark, а сам Spark — один из одновременно зрелых и активно развивающихся проектов Hadoop. Оркестрация и конфигурация потоков была делегирована Apache Airflow, это стандарт нашей фабрики. Для хранения credentials выбрали Hashicorp Vault, что позволило избежать уязвимостей Airflow с точки зрения безопасности.  

Метаинформацию решили класть в отдельную БД на Pоstgres, чтобы не заставлять себя и hdfs страдать от мелких, часто обновляемых файлов. Результатом такого верхнеуровневого проектирования стал следующий архитектурный дизайн:

Что было дальше

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

Agile-подход помог максимально быстро выйти на MVP и раз в спринт выпускать новую версию, реализующую очередную часть заявленной функциональности. Как и все новые проекты в банке, мы используем стандартный набор инструментов для реализации CI/CD. Регулярная обратная связь от коллег, пользующихся фреймворком, позволила сохранить баланс между покрытием legacy-алгоритмов и возрастающей сложностью настройки. В ходе разработки пришлось решать ряд инженерных вопросов, например, compaction-файлов в hdfs и запрет параллельной записи в одну таблицу.

В процессе проектирования мы также решили оптимизировать процесс хранения мелких файлов. Для оптимального хранения, как известно, желательно в hdfs-директории складывать файлы размером около hdfs block size каждый. Для этого перед записью файлов нужно знать размер датафрейма, пришедшего на шаге Extract. Здесь мы решили поэкспериментировать и взяли из текущего кластера информацию по каждой таблице об объеме памяти, которую она занимает в hdfs, о количестве строк и о типах всех атрибутов в данной таблице, чтобы вычислить зависимость между количеством строк, столбцов, типов данных и целевой переменной — размером датасета.

Интерес представляет случай, когда статус объекта — RUNNING. Тут, несмотря на статус, возможны два случая: 

1) Объект действительно сейчас грузится.

2) Кластер упал во время работы приложения, и терминальный статус по объекту уже некому было проставить.

Для определения второго случая мы из приложения идем в логи yarn и по applicationId определяем статус последнего приложения. Оно либо RUNNING, либо KILLED. И мы предпринимаем skip\sleep, либо disaster recovery plan соответственно.

Спустя три спринта разработки команда сформировала продукт с широким охватом функциональности и достаточной стабильностью. На этом прочном фундаменте стало возможно с минимальными трудозатратами проводить тестирования новых lakehouse-фреймворков работы с объектами на уровне hdfs.

Как известно, к Hdfs и Hive применим принцип write once read many. Что на уровне разработки вызывает неудобства и различные самописные реализации обновления данных в таблицах. Эту задачу решили в четырех разных фреймворках: Delta Lake, Hudi, Iceberg, Carbondata. В рамках пилотирования мы сделали четыре ветки для работы с каждым из фреймворков, что позволило провести сравнение производительности, оптимальности и зрелости фреймворков. Выбранный на основе критериальной таблицы лидер был влит в мастер-ветку проекта в виде отдельного алгоритма загрузки.

Помимо функциональности, несущей полезность для нас как для команды, загружающего данные в Data Lake, метаинформация о ETL-потоках позволяет потребителям наших данных оптимизировать свои процессы. Самый очевидный кейс такого использования — проверка готовности таблиц перед построением витрин над ними. Также business critical процессы, пользующиеся нашими данными, на основе метаинформации способны нивелировать временной промежуток в цепочке процессов с помощью событийных механизмов.

Чего мы добились?

• Результатом работы стал общий движок загрузки в data lake c обвязкой отдельных полезных сервисов (рис. 3);

• За квартал тиражные команды смогли настроить загрузку более 2000 объектов. Ежедневный поток обрабатываемых данных — 2.2 Тб;

• Снизился порог вхождения. Для настройки потока не требуется знаний языков программирования;

• Создали общую базу кода на промышленном контуре;

• Упростили процесс миграции с Cloudera Hadoop на Arenadata Hadoop;

• Сократили время загрузки initial load с помощью автоматического режима многопоточности.

Но все это лишь первая версия продукта, наша ближайшая цель  — его декомпозиция в полноценный фреймворк.