company_banner

Из нагруженной MPP СУБД — бодрый Data Lake с аналитическими инструментами: делимся подробностями создания

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



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

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

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

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

    Примерно к такой ситуации мы пришли два-три года назад. На тот момент у нас имелось хранилище на базе MPP СУБД Teradata с использованием ELT-инструмента SAS Data Integration Studio. Это хранилище мы строили с 2011 года вместе с компанией Glowbyte Consulting. В него интегрировали более 15 крупных банковских систем и при этом накопили достаточный объем данных для внедрения и развития аналитических приложений. Кстати, как раз в это время объем данных в основных слоях хранилища из-за множества разных задач начал расти нелинейно, а продвинутая клиентская аналитика стала одним из основных направлений развития банка. Да и наши дата-сайентисты горели желанием поддержать ее. В общем, для построения Data Research Platform звезды сложились как надо.

    Планируем решение


    Здесь надо пояснить: промышленные ПО и сервера – дорогое удовольствие даже для крупного банка. Далеко не каждая организация может позволить себе хранение большого объема данных в топовых MPP СУБД. Всегда приходится делать выбор между ценой и скоростью, надежностью и объемом.

    Чтобы по максимуму использовать имеющиеся возможности, решили поступить вот так:

    • ELT-нагрузку и наиболее востребованную часть исторических данных ХД оставить на СУБД Teradata;
    • полную историю отгружать в Hadoop, позволяющий хранить информацию значительно дешевле.

    Примерно в то время экосистема Hadoop стала не только модной, но и достаточной надежной, удобной для enterprise-применения. Нужно было выбрать дистрибутив. Можно было собрать свой собственный или использовать открытый Apache Hadoop. Но среди enterprise-решений на базе Hadoop себя больше зарекомендовали готовые дистрибутивы от других вендоров — Cloudera и Hortonworks. Поэтому мы тоже решили использовать готовый дистрибутив.

    Так как нашей основной задачей было все-таки хранение структурированных больших данных, то в стеке Hadoop нас интересовали решения, максимально близкие к классическим SQL СУБД. Лидерами здесь являются Impala и Hive. Cloudera развивает и интегрирует решения Impala, Hortonworks – Hive.

    Для углубленного исследования мы организовали для обеих СУБД нагрузочное тестирование, учитывающее профильную для нас нагрузку. Надо сказать, что движки обработки данных в Impala и Hive существенно отличаются — Hive вообще представляет несколько разных вариантов. Однако выбор пал на Impala – и, соответственно, дистрибутив от Cloudera.

    Чем понравилась Impala


    • Высокая скорость выполнения аналитических запросов за счет альтернативного подхода по отношению к MapReduce. Промежуточные результаты вычислений не скидываются в HDFS, что существенно ускоряет обработку данных.
    • Эффективная работа с поколоночным хранением данных в Parquet. Для аналитических задач часто используются так называемые широкие таблицы с множеством колонок. Все колонки используются редко — возможность поднимать из HDFS только нужные для работы позволяет экономить оперативную память и значительно ускорять запрос.
    • Изящное решение с runtime-фильтрами, включающими bloom-фильтрацию. И Hive, и Impala существенно ограничены в использовании индексов, обычных для классических СУБД — из-за особенностей файловой системы хранения HDFS. Поэтому для оптимизации исполнения SQL-запроса движок СУБД должен эффективно воспользоваться доступным партиционированием даже когда оно не задано явно в условиях запроса. Кроме того, ему нужно попытаться предсказать, какое минимальное количество данных из HDFS нужно поднять для гарантированной обработки всех строк. В Impala это работает очень хорошо.
    • Impala использует LLVM – компилятор на виртуальной машине с RISC-подобными инструкциями – для генерации оптимального кода выполнения SQL-запроса.
    • Поддерживаются ODBC и JDBC интерфейсы. Это позволяет почти из коробки интегрировать данные Impala с аналитическими инструментами и приложениями.
    • Есть возможность использования Kudu – чтобы обойти часть ограничений HDFS, и, в частности, писать конструкции UPDATE и DELETE в SQL-запросах.

    Sqoop и остальная архитектура


    Следующим по важности инструментом в стеке Hadoop стал для нас Sqoop. Он позволяет перебрасывать данные между реляционными СУБД (нас, конечно, интересовала Teradata) и HDFS в Hadoop-кластере в разных форматах, включая Parquet. В тестах Sqoop показал высокую гибкость и производительность, поэтому мы решили воспользоваться им — вместо разработки собственных инструментов захвата данных через ODBC/JDBC и сохранения в HDFS.

    Для обучения моделей и смежных задач Data Science, которые удобнее выполнять прямо на кластере Hadoop, мы использовали Spark от Apache. В своей области он стал стандартным решением — и есть за что:

    • библиотеки машинного обучения Spark ML;
    • поддержка четырех языков программирования (Scala, Java, Python, R);
    • интеграция с аналитическими инструментами;
    • in-memory обработка данных дает отличную производительность.

    В качестве аппаратной платформы закупили сервера Oracle Big Data Appliance. Начинали с шести узлов в продуктивном контуре с 2x24-core CPU и 256 ГБ памяти на каждом. Текущая конфигурация содержит 18 таких же узлов с расширенной до 512 ГБ памятью.



    На схеме показана верхнеуровневая архитектура Data Research Platform и смежных систем. Центральное звено — кластер Hadoop на базе дистрибутива Cloudera (CDH). Он используется для как для получения с помощью Sqoop и хранения данных КХД в HDFS — в поколоночном формате Parquet, допускающем использование кодеков для сжатия, например, Snappy. Также кластер обрабатывает данные: Impala используется для ELT-подобных трансформаций, Spark — для Data Science задач. Для разделения доступа к данным используется Sentry.

    Impala имеет интерфейсы для практически всех современных enterprise-средств аналитики. Помимо этого в качестве клиентов могут подключаться произвольные инструменты, поддерживающие ODBC/JDBC интерфейсы. Для работы с SQL мы в качестве основных клиентов рассматриваем Hue и TOAD for Hadoop.

    Для управления всеми потоками, которые указаны на схеме стрелочками, предназначена ETL-подсистема, состоящая из средств SAS (Metadata Server, Data Integration Studio), и ETL фреймворк, написанный на базе SAS и shell-скриптов с использованием базы для хранения метаданных ETL-процессов. Руководствуясь правилами, заданными в метаданных, ETL-подсистема запускает процессы обработки данных как на КХД, так и на Data Research Platform. В итоге мы имеем сквозную систему мониторинга и управления потоками данных вне зависимости от используемой среды (Teradata, Impala, Spark и прочее, если в том будет потребность).

    Через грабли к звездам


    Разгрузить КХД вроде бы просто. На входе и выходе реляционные СУБД, бери да переливай данные через Sqoop. Судя по описанию выше, у нас все шло очень гладко, но, конечно же, без приключений не обошлось, и это, пожалуй, самая интересная часть всего проекта.



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

    Sqoop – очень качественный инструмент, но не во всех случаях и комбинациях систем он работает абсолютно надежно. На наших объемах недостаточно оптимально работал коннектор к Teradata. Мы воспользовались открытостью кода Sqoop и внесли изменения в библиотеки коннектора. Стабильность соединения при перемещении данных увеличилась.

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

    Все MPP, и Teradata в частности, обладают особенностью, связанной с параллельным хранением данных и исполнением инструкций. Если эту особенность не принимать в расчет, то может оказаться, что всю работу возьмет на себя один логический узел кластера, из-за чего выполнение запроса станет гораздо медленнее, раз этак в 100-200. Мы, конечно, не могли этого допустить, поэтому написали специальный движок, который использует ETL-метаданные таблиц КХД и выбирает оптимальную степень параллелизации задач Sqoop.

    Историчность в хранилище – дело тонкое, особенно если использовать SCD2, при том что в Impala не поддерживаются UPDATE и DELETE. Мы, конечно, хотим, чтобы исторические таблицы в Data Research Platform выглядели абсолютно так же, как в Teradata. Этого можно достигнуть, комбинируя получение инкремента через Sqoop, выделение обновляемых бизнес-ключей и удаление партиций в Impala. Чтобы эту вычурную логику не пришлось писать каждому разработчику, мы упаковали ее в специальную библиотеку (на нашем ETL-сленге «загрузчик»).

    Напоследок — вопрос с типами данных. Impala достаточно свободно относится к конвертации типов, поэтому с какими-то затруднениями мы столкнулись только в типах TIMESTAMP и CHAR/VARCHAR. Для даты-времени мы решили хранить данные в Impala в текстовом (STRING) формате YYYY-MM-DD HH:MM:SS. Этот подход, как оказалось, вполне позволяет использовать функции трансформации даты и времени. Для строковых данных заданной длины оказалось, что хранение в формате STRING в Impala ничем им не уступает, поэтому мы тоже использовали его.

    Обычно для организации Data Lake копируют данные источников в полуструктурированных форматах в специальную stage-область в Hadoop, после чего средствами Hive или Impala устанавливают схему десериализации этих данных для использования их в SQL-запросах. Мы пошли по тому же пути. Важно отметить, что не все и не всегда имеет смысл тащить в хранилище данных, так как разработка процессов копирования файликов и установка схемы значительно дешевле загрузки бизнес-атрибутов в модель КХД с помощью ETL-процессов. Когда еще непонятно, в каком объеме, на какой срок и с какой периодичностью нужны данные источника, Data Lake в описанном подходе является простым и дешевым решением. Сейчас мы регулярно загружаем в Data Lake прежде всего источники, генерирующие пользовательские события: данные анализа заявок, логи и сценарии перехода автозвонилки и автоответчика Avaya, карточные транзакции.

    Инструментарий аналитиков


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

    • Удобство инструмента в использовании и поддержке
    • Применимость в задачах Data Science
    • Максимальная возможность использования вычислительных ресурсов кластера Hadoop, а не серверов приложений или компьютера исследователя

    И вот на чем остановились:

    • Python + Anaconda. В качестве среды используется iPython/Jupyter
    • R + Shiny. Исследователь работает в desktop или web-версии R Studio, Shiny применяется для разработки web-приложений, которые заточены на использование алгоритмов, разработанных в R.
    • Spark. Для работы с данными используются интерфейсы для Python (pyspark) и R, настроенные в средах разработки, указанных в предыдущих пунктах. Оба интерфейса позволяют задействовать библиотеку Spark ML, которая дает возможность обучать ML-модели на кластере Hadoop/Spark.
    • Данные в Impala доступны через Hue, Spark и из сред разработки с помощью стандартного интерфейса ODBC и специальных библиотек типа implyr

    Сейчас в Data Lake находится порядка 100 Тб данных из розничного хранилища плюс около 50 Тб из ряда OLTP-источников. Озеро обновляется ежедневно в инкрементальном режиме. В дальнейшем мы собираемся повышать удобство для пользователей, выводить ELT-нагрузку на Impala, увеличивать количество источников, загружаемых в Data Lake, и расширять возможности для продвинутой аналитики.

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

    • Используйте лучшие практики. Если бы мы не имели ETL-подсистемы, метаданных, версионного хранения и понятной архитектуры, то не осилили бы эту задачу. Лучшие практики окупают себя, хотя и не сразу.
    • Помните об объемах данных. Большие данные могут создавать технические сложности в совсем неожиданных местах.
    • Следите за новыми технологиями. Новые решения появляются часто, не все они полезны, но иногда встречаются настоящие жемчужины.
    • Больше экспериментируйте. Не стоит доверять только маркетинговым описаниям решений – пробуйте сами.

    Кстати, о том, как наши аналитики использовали машинное обучение и данные банка для работы с кредитными рисками, вы можете почитать в отдельном посте.
    ВТБ
    Company

    Comments 7

      0
      В качестве аппаратной платформы закупили сервера Oracle Big Data Appliance. Начинали с шести узлов в продуктивном контуре с 2x24-core CPU и 256 ГБ памяти на каждом. Текущая конфигурация содержит 18 таких же узлов с расширенной до 512 ГБ памятью.

      Сейчас в Data Lake находится порядка 100 Тб данных из розничного хранилища плюс около 50 Тб из ряда OLTP-источников.


      Интересно было бы взглянуть на тесты производительности, т.к. объем данных вызывает улыбку на фоне доступных ресурсов

        0
        Задача хранилища данных не сводится к накоплению максимального их объема. Нужно еще максимизировать конкурентность и производительность пользовательских и ELT-запросов, надежность системы и при этом минимизировать стоимость владения. Также можно уточнить, что в Hadoop данные можно хранить с использованием компрессии (например, Snappy). Если имеющиеся у нас в DRP данные представить в виде CSV без компрессии, то их объем был бы около 900 Тб. Без учета этих факторов хранилища на классических MPP тоже могут показаться недостаточно большими, хотя они и будут обеспечивать хорошую производительность. На текущий момент, вы абсолютно правы, у нас действительно серьезные задел по производительности.
        0
        Спасибо!
        Управление проектами стандартный РМ или аджайл?
          0
          К управлению проектами и задачами стараемся подходить гибко. Описанный функционал создавался в рамках нескольких проектов и задач, подходы к управлению применялись различные.
          0
          Как-то у вас про импалу уж очень оптимистично все. Неужели например на issues.apache.org/jira/browse/IMPALA-3316 не натыкались? (да, исправили, да, через два года всего-навсего ;)

          У нас как-то почти не видно таких, что был бы в особом восторге — во-первых, несовместимо, то есть к несовместимостям скажем Spark и Hive добавляются лишние проблемы между Hive и Impala. Во-вторых, со сложными типами (array, struct и map) все тоже как-то как минимум неудобно — а при отсутствии индексов де-факто эти типы и денормализация местами наше все.

          >Данные в Impala доступны через Hue, Spark

          Это вы что имели в виду, когда говорили про Spark? У импалы разве есть какие-то «свои» данные, с которыми может работать Spark?
            0
            1) Да, мы знаем про ограничения использования TIMESTAMP и поэтому используем представление даты-времени в STRING, как это указано в статье. Сама проблема заключается в том, что если писать в Parquet поля TIMESTAMP с помощью Hive, а потом читать Impala, то может возникнуть серьезное замедление запроса.
            Но это не наш случай, так как мы не используем Hive.

            2) Мы стараемся выбирать способы использования, для которых совместимость и работоспособность гарантируется. Практически во всех СУБД и особенно MPP-СУБД есть свои ограничения и вопрос только в том, умеет ли DWH-команда находить приемлемые способы борьбы с ними. Мы — нашли. Ну и задачи DWH пока не потребовали у нас хранения сложных типов данных — мы обходимся простыми.

            С точки зрения производительности отсутствие индексов не сильно влияет на нас из-за наличия runtime bloom-фильтров в Impala. Также для нас важна не производительность на единичных запросах, а в условиях их конкурентности и ad hoc + ELT нагрузки. По нашим замерам, Impala не уступает в этом профиле нагрузки классическим MPP.
            OLTP-like нагрузка это немного другой профиль и он тоже может быть интересен. Тут есть несколько решений: во-первых, первичные ключи в Kudu отчасти похожи на индексы и для ряда приложений мы исследуем возможность хранения данных в Kudu; во-вторых мы ждем реализации доработок Impala, которые также должны увеличить производительность для немассовых выборок данных (https://blog.cloudera.com/blog/2017/12/faster-performance-for-selective-queries/ и issues.apache.org/jira/browse/IMPALA-3430)

            3) Имелось в виду обращение Spark к данным в HDFS через схемы Hive Metastore, который использует и Impala. Impala не имеет «своих» данных (если опустить оговорку про Kudu) и является SQL-движком для параллельного выполнения на кластере.
            0
            1. Toad for hadoop более не доступен, его лицензия была не совсем free.
            2. Два года назад банк в котором использовалось данное решение назывался немного по другому.

            Only users with full accounts can post comments. Log in, please.