Когда у вас сберовские масштабы. Использование Ab Initio при работе с Hive и GreenPlum

Некоторое время назад перед нами встал вопрос выбора ETL-средства для работы с BigData. Ранее использовавшееся решение Informatica BDM не устраивало нас из-за ограниченной функциональности. Её использование свелось к фреймворку по запуску команд spark-submit. На рынке имелось не так много аналогов, в принципе способных работать с тем объёмом данных, с которым мы имеем дело каждый день. В итоге мы выбрали Ab Initio. В ходе пилотных демонстраций продукт показал очень высокую скорость обработки данных. Информации об Ab Initio на русском языке почти нет, поэтому мы решили рассказать о своём опыте на Хабре.

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

Он помогает бизнесу глобально копить знания и развивать экосистему, а разработчику — прокачивать свои навыки в ETL, подтягивать знания в shell, предоставляет возможность освоения языка PDL, даёт визуальную картину процессов загрузки, упрощает разработку благодаря обилию функциональных компонентов.

В посте я расскажу о возможностях Ab Initio и приведу сравнительные характеристики по его работе с Hive и GreenPlum.

• Описание фреймворка MDW и работ по его донастройке под GreenPlum
• Сравнительные характеристики производительности Ab Initio по работе с Hive и GreenPlum
• Работа Ab Initio с GreenPlum в режиме Near Real Time

Функционал этого продукта очень широк и требует немало времени на своё изучение. Однако, при должных навыках работы и правильных настройках производительности результаты обработки данных получаются весьма впечатляющие. Использование Ab Initio для разработчика может дать ему интересный опыт. Это новый взгляд на ETL-разработку, гибрид между визуальной средой и разработкой загрузок на скрипто-подобном языке.

Бизнес развивает свои экосистемы и этот инструмент оказывается ему как никогда кстати. С помощью Ab Initio можно копить знания о текущем бизнесе и использовать эти знания для расширения старых и открытия новых бизнесов. Альтернативами Ab Initio можно назвать из визуальных сред разработки Informatica BDM и из невизуальных сред – Apache Spark.

Описание Ab Initio

Ab Initio, как и другие ETL-средства, представляет собой набор продуктов.

Ab Initio GDE (Graphical Development Environment) – это среда для разработчика, в которой он настраивает трансформации данных и соединяет их потоками данных в виде стрелочек. При этом такой набор трансформаций называется графом:

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

Имеется несколько сотен функциональных компонентов, что очень много. Многие из них узкоспециализированные. Возможности классических трансформаций в Ab Initio шире, чем в других ETL-средствах. Например, Join имеет несколько выходов. Помимо результата соединения датасетов можно получить на выходе записи входных датасетов, по ключам которых не удалось соединиться. Также можно получить rejects, errors и лог работы трансформации, который можно в этом же графе прочитать как текстовый файл и обработать другими трансформациями:

Или, например, можно материализовать приёмник данных в виде таблицы и в этом же графе считать из него данные.

Есть оригинальные трансформации. Например, трансформация Scan имеет функционал, как у аналитических функций. Есть трансформации с говорящими названиями: Create Data, Read Excel, Normalize, Sort within Groups, Run Program, Run SQL, Join with DB и др. Графы могут использовать параметры времени выполнения, в том числе возможна передача параметров из операционной системы или в операционную систему. Файлы с готовым набором передаваемых графу параметров называются parameter sets (psets).

Как и полагается, Ab Initio GDE имеет свой репозиторий, именуемый EME (Enterprise Meta Environment). Разработчики имеют возможность работать с локальными версиями кода и делать check in своих разработок в центральный репозиторий.

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

Также есть возможность кликнуть по любому потоку и посмотреть tracking details – в сколько параллелей работала трансформация, сколько строк и байт в какой из параллелей загрузилось:

Есть возможность разбить выполнение графа на фазы и пометить, что одни трансформации нужно выполнять первым делом (в нулевой фазе), следующие в первой фазе, следующие во второй фазе и т.д.

У каждой трансформации можно выбрать так называемый layout (где она будет выполняться): без параллелей или в параллельных потоках, число которых можно задать. При этом временные файлы, которые создаёт Ab Initio при работе трансформаций, можно размещать как в файловой системе сервера, так и в HDFS.

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

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

Также в Ab Initio хорошо развита интеграция с ОС через shell. Конкретно в Сбербанке используется linux ksh. Можно обмениваться с shell переменными и использовать их в качестве параметров графов. Можно из shell вызывать выполнение графов Ab Initio и администрировать Ab Initio.

Помимо Ab Initio GDE в поставку входит много других продуктов. Есть своя Co>Operation System с претензией называться операционной системой. Есть Control>Center, в котором можно ставить на расписание и мониторить потоки загрузки. Есть продукты для осуществления разработки на более примитивном уровне, чем позволяет Ab Initio GDE.

Описание фреймворка MDW и работ по его донастройке под GreenPlum

Вместе со своими продуктами вендор поставляет продукт MDW (Metadata Driven Warehouse), который представляет собой конфигуратор графов, предназначенный для помощи в типичных задачах по наполнению хранилищ данных или data vaults.

Он содержит пользовательские (специфичные для проекта) парсеры метаданных и готовые генераторы кода “из коробки”.

На входе MDW получает модель данных, конфигурационный файл по настройке соединения с базой данных (Oracle, Teradata или Hive) и некоторые другие настройки. Специфическая для проекта часть, например, разворачивает модель в базе данных. Коробочная часть продукта генерирует графы и настроечные файлы к ним по загрузке данных в таблицы модели. При этом создаются графы (и psets) для нескольких режимов инициализирующей и инкрементальной работы по обновлению сущностей.

В случаях Hive и RDBMS генерируются различающиеся графы по инициализирующему и инкрементальному обновлению данных.

В случае Hive поступившие данные дельты соединяется посредством Ab Initio Join с данными, которые были в таблице до обновления. Загрузчики данных в MDW (как в Hive, так и в RDBMS) не только вставляют новые данные из дельты, но и закрывают периоды актуальности данных, по первичным ключам которых поступила дельта. Кроме того, приходится переписать заново неизменившуюся часть данных. Но так приходится делать, поскольку в Hive нет операций delete или update.

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

Поступившая дельта загружается в промежуточную таблицу в базу данных. После этого происходит соединение дельты с данными, которые были в таблице до обновления. И делается это силами SQL посредством сгенерированного SQL-запроса. Далее с помощью SQL-команд delete+insert в целевую таблицу происходит вставка новых данных из дельты и закрытие периодов актуальности данных, по первичным ключам которых поступила дельта.
Неизменившиеся данные переписывать нет нужды.

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

Для проекта в Сбербанке мы создали новую многократно используемую реализацию загрузчика базы данных для GreenPlum. Сделано это было на основе версии, которую MDW генерирует для Teradata. Именно Teradata, а не Oracle подошла для этого лучше и ближе всего, т.к. тоже является MPP-системой. Способы работы, а также синтаксис Teradata и GreenPlum оказались близки.

Примеры критичных для MDW различий между разными RDBMS таковы. В GreenPlum в отличии от Teradata при создании таблиц нужно писать клаузу

distributed by

В Teradata пишут

delete <table> all

, а в GreеnPlum пишут

delete from <table>

В Oracle в целях оптимизации пишут

delete from t where rowid in (<соединение t с дельтой>)

, а в Teradata и GreenPlum пишут

delete from t where exists (select * from delta where delta.pk=t.pk)

Ещё отметим, что для работы Ab Initio с GreenPlum потребовалось установить клиент GreenPlum на все ноды кластера Ab Initio. Это потому, что мы подключились к GreenPlum одновременно со всех узлов нашего кластера. А для того, чтобы чтение из GreenPlum было параллельным и каждый параллельный поток Ab Initio читал свою порцию данных из GreenPlum, пришлось в секцию «where» SQL-запросов поместить понимаемую Ab Initio конструкцию

where ABLOCAL()

и определить значение этой конструкции, указав читающей из БД трансформации параметр

ablocal_expr=«string_concat("mod(t.", string_filter_out("{$TABLE_KEY}","{}"), ",", (decimal(3))(number_of_partitions()),")=", (decimal(3))(this_partition()))»

, которая компилируется в что-то типа

mod(sk,10)=3

, т.е. приходится подсказывать GreenPlum явный фильтр для каждой партиции. Для других баз данных (Teradata, Oracle) Ab Initio может выполнить это распараллеливание автоматически.

Сравнительные характеристики производительности Ab Initio по работе с Hive и GreenPlum

В Сбербанке был проведён эксперимент по сравнению производительности сгенерированных MDW графов применительно к Hive и применительно к GreenPlum. В рамках эксперимента в случае Hive имелось 5 нод на том же кластере, что и Ab Initio, а в случае GreenPlum имелось 4 ноды на отдельном кластере. Т.е. Hive имел некоторое преимущество над GreenPlum «по железу».

Было рассмотрено две пары графов, выполняющих одну и ту же задачу обновления данных в Hive и в GreenPlum. При этом запускали графы, сгенерированные конфигуратором MDW:

• инициализирующая загрузка + инкрементальная загрузка случайно сгенерированных данных в таблицу Hive
• инициализирующая загрузка + инкрементальная загрузка случайно сгенерированных данных в такую же таблицу GreenPlum

В обоих случаях (Hive и GreenPlum) запускали загрузки в 10 параллельных потоков на одном и том же кластере Ab Initio. Промежуточные данные для расчётов Ab Initio сохранял в HDFS (в терминах Ab Initio был использован MFS layout using HDFS). Одна строка случайно сгенерированных данных занимала в обоих случаях по 200 байт.

Результат получился такой:

Hive:

Инициализирующая загрузка в Hive

Вставлено строк
6 000 000
60 000 000
600 000 000

Продолжительность инициализирующей
загрузки в секундах
41
203
1 601

Инкрементальная загрузка в Hive

Количество строк, имевшихся в
целевой таблице на начало эксперимента
6 000 000
60 000 000
600 000 000

Количество строк дельты, применённых к
целевой таблице в ходе эксперимента
6 000 000
6 000 000
6 000 000

Продолжительность инкрементальной
загрузки в секундах
88
299
2 541

GreenPlum:

Инициализирующая загрузка в GreenPlum

Вставлено строк
6 000 000
60 000 000
600 000 000

Продолжительность инициализирующей
загрузки в секундах
72
360
3 631

Инкрементальная загрузка в GreenPlum

Количество строк, имевшихся в
целевой таблице на начало эксперимента
6 000 000
60 000 000
600 000 000

Количество строк дельты, применённых к
целевой таблице в ходе эксперимента
6 000 000
6 000 000
6 000 000

Продолжительность инкрементальной
загрузки в секундах
159
199
321

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

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

Инкрементальная же загрузка в GreenPlum слабо зависит от объёма имеющихся в целевой таблице ранее загруженных данных и проходит достаточно быстро. Получилось это благодаря SQL Joins и архитектуре GreenPlum, допускающей операцию delete.

Итак, GreenPlum вливает дельту методом delete+insert, а в Hive нету операций delete либо update, поэтому весь массив данных при инкрементальном обновлении были вынуждены переписывать целиком. Наиболее показательно сравнение выделенных жирным ячеек, так как оно соответствует наиболее частому варианту эксплуатации ресурсоёмких загрузок. Видим, что GreenPlum выиграл у Hive в этом тесте в 8 раз.

Работа Ab Initio с GreenPlum в режиме Near Real Time

В этом эксперименте проверим возможность Ab Initio производить обновление таблицы GreenPlum случайно формируемыми порциями данных в режиме, близком к реальному времени. Рассмотрим таблицу GreenPlum dev42_1_db_usl.TESTING_SUBJ_org_finval, с которой будет вестись работа.

Будем использовать три графа Ab Initio по работе с ней:

1) Граф Create_test_data.mp – создаёт в 10 параллельных потоков файлы с данными в HDFS на 6 000 000 строк. Данные случайные, структура их организована для вставки в нашу таблицу

2) Граф mdw_load.day_one.current.dev42_1_db_usl_testing_subj_org_finval.pset – сгенерированный MDW граф по инициализирующей вставке данных в нашу таблицу в 10 параллельных потоков (используются тестовые данные, сгенерированные графом (1))

3) Граф mdw_load.regular.current.dev42_1_db_usl_testing_subj_org_finval.pset – сгенерированный MDW граф по инкрементальному обновлению нашей таблицы в 10 параллельных потоков с использованием порции свежих поступивших данных (дельты), сгенерированных графом (1)

Выполним нижеприведённый сценарий в режиме NRT:

• сгенерировать 6 000 000 тестовых строк
• произвести инициализирующую загрузку вставить 6 000 000 тестовых строк в пустую таблицу
• повторить 5 раз инкрементальную загрузку
  • сгенерировать 6 000 000 тестовых строк
  • произвести инкрементальную вставку 6 000 000 тестовых строк в таблицу (при этом старым данным проставляется время истечения актуальности valid_to_ts и вставляются более свежие данные с тем же первичным ключом)

Такой сценарий эмулирует режим реальной работы некой бизнес-системы – в режиме реального времени появляется достаточно объёмная порция новых данных и тут же вливается в GreenPlum.

Теперь посмотрим лог работы сценария:

Start Create_test_data.input.pset at 2020-06-04 11:49:11
Finish Create_test_data.input.pset at 2020-06-04 11:49:37
Start mdw_load.day_one.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 11:49:37
Finish mdw_load.day_one.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 11:50:42
Start Create_test_data.input.pset at 2020-06-04 11:50:42
Finish Create_test_data.input.pset at 2020-06-04 11:51:06
Start mdw_load.regular.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 11:51:06
Finish mdw_load.regular.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 11:53:41
Start Create_test_data.input.pset at 2020-06-04 11:53:41
Finish Create_test_data.input.pset at 2020-06-04 11:54:04
Start mdw_load.regular.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 11:54:04
Finish mdw_load.regular.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 11:56:51
Start Create_test_data.input.pset at 2020-06-04 11:56:51
Finish Create_test_data.input.pset at 2020-06-04 11:57:14
Start mdw_load.regular.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 11:57:14
Finish mdw_load.regular.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 11:59:55
Start Create_test_data.input.pset at 2020-06-04 11:59:55
Finish Create_test_data.input.pset at 2020-06-04 12:00:23
Start mdw_load.regular.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 12:00:23
Finish mdw_load.regular.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 12:03:23
Start Create_test_data.input.pset at 2020-06-04 12:03:23
Finish Create_test_data.input.pset at 2020-06-04 12:03:49
Start mdw_load.regular.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 12:03:49
Finish mdw_load.regular.current.dev42_1_db_usl_testing_subj_org_finval.pset at 2020-06-04 12:06:46

Получается такая картина:

Graph
Start time
Finish time
Length

Create_test_data.input.pset
04.06.2020 11:49:11
04.06.2020 11:49:37
00:00:26

mdw_load.day_one.current.
dev42_1_db_usl_testing_subj_org_finval.pset
04.06.2020 11:49:37
04.06.2020 11:50:42
00:01:05

Create_test_data.input.pset
04.06.2020 11:50:42
04.06.2020 11:51:06
00:00:24

mdw_load.regular.current.
dev42_1_db_usl_testing_subj_org_finval.pset
04.06.2020 11:51:06
04.06.2020 11:53:41
00:02:35

Create_test_data.input.pset
04.06.2020 11:53:41
04.06.2020 11:54:04
00:00:23

mdw_load.regular.current.
dev42_1_db_usl_testing_subj_org_finval.pset
04.06.2020 11:54:04
04.06.2020 11:56:51
00:02:47

Create_test_data.input.pset
04.06.2020 11:56:51
04.06.2020 11:57:14
00:00:23

mdw_load.regular.current.
dev42_1_db_usl_testing_subj_org_finval.pset
04.06.2020 11:57:14
04.06.2020 11:59:55
00:02:41

Create_test_data.input.pset
04.06.2020 11:59:55
04.06.2020 12:00:23
00:00:28

mdw_load.regular.current.
dev42_1_db_usl_testing_subj_org_finval.pset
04.06.2020 12:00:23
04.06.2020 12:03:23
00:03:00

Create_test_data.input.pset
04.06.2020 12:03:23
04.06.2020 12:03:49
00:00:26

mdw_load.regular.current.
dev42_1_db_usl_testing_subj_org_finval.pset
04.06.2020 12:03:49
04.06.2020 12:06:46
00:02:57

Видим, что 6 000 000 строк инкремента обрабатываются за 3 минуты, что достаточно быстро.
Данные в целевой таблице получились распределёнными следующим образом:

select valid_from_ts, valid_to_ts, count(1), min(sk), max(sk) from dev42_1_db_usl.TESTING_SUBJ_org_finval group by valid_from_ts, valid_to_ts order by 1,2;

Можно разглядеть соответствие вставленных данных моментам запуска графов.
Значит можно запускать в Ab Initio инкрементальную загрузку данных в GreenPlum с очень высокой частотой и наблюдать высокую скорость вставки этих данных в GreenPlum. Конечно, раз в секунду запускаться не получится, так как Ab Initio, как и любое ETL-средство, при запуске требует времени «на раскачку».

Заключение

Сейчас Ab Initio используется в Сбербанке для построения Единого семантического слоя данных (ЕСС). Этот проект подразумевает построение единой версии состояния различных банковских бизнес-сущностей. Информация приходит из различных источников, реплики которых готовятся на Hadoop. Исходя из потребностей бизнеса, готовится модель данных и описываются трансформации данных. Ab Initio загружает информацию в ЕСС и загруженные данные не только представляют интерес для бизнеса сами по себе, но и служат источником для построения витрин данных. При этом функционал продукта позволяет использовать в качестве приёмника различные системы (Hive, Greenplum, Teradata, Oracle), что даёт возможность без особых усилий подготавливать данные для бизнеса в различных требуемых ему форматах.

Возможности Ab Initio широки, например, прилагающийся фреймворк MDW даёт возможность строить техническую и бизнес-историчность данных “из коробки”. Для разработчиков Ab Initio даёт возможность “не изобретать велосипед”, а пользоваться множеством имеющихся функциональных компонентов, по сути являющихся библиотеками, нужными при работе с данными.

Автор — эксперт профессионального сообщества Сбербанка SberProfi DWH/BigData. Профессиональное сообщество SberProfi DWH/BigData отвечает за развитие компетенций в таких направлениях, как экосистема Hadoop, Teradata, Oracle DB, GreenPlum, а также BI инструментах Qlik, SAP BO, Tableau и др.

Источник: habr.com