Mass parallel requests - Greenplum
Last updated
Last updated
https://habr.com/ru/companies/tinkoff/articles/694652/
Меня зовут Дмитрий Немчин, я руковожу отделом, который отвечает за движки хранения и обработки данных в платформе данных Тинькофф. Несколько лет назад мы поняли, что продукты, на которых работало хранилище, перестали нас устраивать. Объемы росли, понадобилось масштабируемое решение. В этом тексте я расскажу, как мы пришли к Greenplum в качестве ядра хранилища данных и как используем его.
Вообще, эта статья должна была стать расшифровкой моего доклада, с которым я выступал на HighLoad++ в 2018 году. Но с тех пор у нас произошло много изменений, которые очень хочется показать сообществу. Поэтому статья — микс из ретроспективы и текущих реалий нашей большой платформы данных. Кроме того, в процессе у статьи появился второй автор — написать ее помог @koskar
В начале было слово, и этим словом был SAS. Все наше хранилище когда-то строилось на продуктах SAS. Для обработки данных мы использовали один сервер, хранили их на СХД, а процессы строились только на использовании паттерна ETL. Если говорить просто, то: открыли базу-источник, что-то оттуда забрали, преобразовали и положили в КХД.
Все работало отлично, пока не выросли объемы (удивительно, не правда ли?). Конечно же, мы уперлись в процессор с памятью на сервере обработки и производительность СХД.
Нужно было что-то менять, причем радикально. Изучив разные решения, мы выбрали Greenplum за высокую производительность и простоту (слегка обманчивую, но все же). Как выяснилось потом, мы получили неплохой бонус: через несколько лет Greenplum вышел в Open Source. В текущих обстоятельствах это очень сильное конкурентное преимущество!
Коротко о том, как Greenplum обрабатывает пользовательские запросы:
Запрос приходит на мастер.
Запрос отправляется на сегменты (небольшие инстансы Postgres).
Каждый сегмент обрабатывает свою часть данных.
Данные со всех сегментов собираются на мастере, сортируются, агрегируются и возвращаются клиенту.
Подробнее о Greenplum можно узнать из этого текста, а как с ним жить с точки зрения DBA — из этого.
Технология сама по себе — это, конечно, хорошо. Но мы не академики, а практики, и нам важно эту технологию применять и правильно встроить ее в нашу платформу и процессы. Как раз этому будет посвящена большая часть статьи.
Greenplum в нашей компании используется с 2013 года. Я же пришел в компанию в 2015 году, когда вокруг Greenplum уже было построено заметное число процессов и даже один внутренний продукт. Но обо всем по порядку.
В те далекие времена заметную часть данных мы читали напрямую из операционных баз данных, а часть реплицировали непосредственно в КХД, используя Attunity Replicate. Данных в DWH у нас было порядка 20 ТБ, каждую ночь запускалось около 2 тысяч ETL\ELT-джобов, которые строили хранилище. Уже тогда у нас работали два боевых кластера из 16 машин каждый (для катастрофоустойчивости) и тестовый кластер, тоже из 16 машин. Во всех серверах Greenplum были только HDD, и все это жило и даже процветало, то есть вполне справлялось с нагрузками.
Но объемы данных вместе с количеством пользователей платформы постоянно росли, и к 2018 году объем данных в нашем Greenplum превысил 70 ТБ. На тот момент мы сделали свое решение для DR и свою систему репликации, а число ETL/ELT-джобов на каждую ночь выросло до примерно 4000. Причем теперь это были по большей части ELT. Объем каждого из двух боевых кластеров вырос до 36 машин, столько же машин теперь было в тестовом кластере.
Часть данных мы стали хранить на очень быстрых NMVe PCIe картах, часть — на SAS SSD, остальное — на RAID-массивах из HDD. Это дало заметный буст скорости работы с данными. Когда мы это затеяли, даже вендор Greenplum (Pivotal, сейчас — часть VMWare) высказывал сомнения в целесообразности такого решения. А сейчас такие решения стали стандартом построения кластеров Greenplum и других MPP (massive parallel processing) БД.
Поговорим подробнее о пользователях нашей платформы. Тинькофф — data-driven компания, и без данных и цифр у нас решения не принимаются. При этом мы стараемся не увеличивать количество бюрократии для получения доступа к данным. Нет, информационная безопасность у нас есть и не дремлет, но это совсем не такая информационная безопасность, какой ее можно представить. Несмотря на то, что у нас есть банковская лицензия, они всеми силами стараются сделать работу компании безопасной, не создавая лишних препятствий.
Потребители данных у нас работают по двум направлениям: отчетность и аналитика. В 2018 году отчетность строилась в основном в SAP BusinessObjects и Tableau, а ad hoc аналитика была сосредоточена в SAS и Apache Zeppelin. MAU платформы в те времена составляло около 3 тысяч пользователей.
Позже на замену Apache Zeppelin пришел Helicopter, а от SAS и SAP мы постепенно уходим, но это уже выходит за рамки темы статьи. При этом MAU выросло до примерно 6 тысяч пользователей. Да, это MAU платформы данных.
А теперь перейдем к тому, что мы построили вокруг Greenplum за время его использования.
Верхнеуровневая структура хранилища данных на 2018 год
Начнем с репликации данных. ETL — это замечательно. Но с ростом объемов все приходит к тому, что постоянно лезть в операционные базы не нужно. Хотя бы потому, что это создает бешеную нагрузку и можно получить по шапке. Ну и да, можно своей нагрузкой сломать боевой процессинг денег или положить сайт/приложение. Вариантов получить критические нагрузки на боевых системах много.
Поэтому все хранилища любят CDC — change data capture. Это захват изменений в данных и метаданных источников, обычно выполняющийся более-менее в реальном времени. Но для доставки таких изменений в MPP систему, например в Greenplum, данные собираются в батчи, чтобы не убивать MPP движок построчными изменениями. Далее эти батчи изменений применяются в Greenplum (в нашем случае). А потом запускаем свой страшный ELT, обрабатываем данные уже в отдельной от боевых баз системе и спокойно строим аналитическое хранилище, отчеты и все, что придет нам в голову. Например, обучение ML.
Вначале, еще до моего прихода в компанию, Тинькофф для этих целей использовал Attunity Replicate. На каждый Greenplum у нас был свой сервер Attunity. У системы было несколько плюсов, но минусы в итоге перевесили.
Disclaimer: все, относящееся к Attunity Replicate верно для ее версий 2015-2016 годов. Примерно в это время мы уже начали строить свою систему репликации данных.
Плюсы:
— удобная система с функциональным веб-интерфейсом;
— хорошо работает на небольших объемах и при стабильной схеме данных.
Минусы:
— row-by-row. Если апдейт не прошел, система начинает построчно проводить апдейты. Greenplum этого крайне не любит;
— отстаивание репликации при нагрузке на источник;
— изменение DDL требует полной перезагрузки таблицы;
— убивает диски Greenplum, создавая слишком большую нагрузку;
— работает на Windows, а для нас был привычнее Linux.
Пожив какое-то время с Attunity, мы пришли к решению создать собственную CDC-систему. Что мы хотели от нее получить?
Во-первых, снизить нагрузку на приемники. Сделать батчи с регулировкой размера и получить возможность произвольное количество времени ничего не грузить (например, выделить окно для работы ETL/ELT без паразитных нагрузок на кластер).
Во-вторых, автоматически применять DDL. Бывало, что из-за ночных релизов у нас падало что-то важное в репликации и приходилось это чинить, а мы хотели спать по ночам.
В-третьих, нам нужно было легко масштабироваться и не добавлять каждый раз новый сервер Attunity. Хотелось добавить новый приемник в уже готовую систему и жить счастливо.
В-четвертых, мы хотели Linux, поскольку фактически все другие сервисы у нас жили и живут на Linux.
Нам удалось всего этого достичь. Текущее решение — не полностью наш продукт: в начале потока загрузки данных мы используем Oracle Golden Gate. С его помощью мы из нескольких десятков баз Oracle сливаем одну большую базу ODS. Из нее выгружаем данные нашими самописными процессами во временные файлы и применяем их аккуратно во все Greenplum с удобными для нас размерами батчей и частотой их применения. Unload, то есть выгрузка из ODS, делается с помощью нашей самописной библиотеки, а apply сделан на механизмах Greenplum.
Сейчас CDC-система переваривает довольно большие объемы. Только в Greenplum хранится более 150 ТБ (на 2018 год было порядка 20 ТБ) данных с учетом сжатия. Теперь мы загружаем данные не только из Oracle, но и из Postgres. Лаг репликации — до двух часов от попадания строки в источник до попадания в DWH для самых критичных данных. Можно сделать еще меньше, но мы не хотим грузить Greenplum. Также есть возможность отгружать данные в Kafka.
Архитектура нашей CDC-системы
Теперь поговорим о disaster recovery для Greenplum. В 2015 году DR-решений для GP, каким его видели мы, просто не было. Больше того: в таком виде его, наверное, нет и сейчас. Есть много других решений, но нам они не подходят по разным причинам.
Сам по себе Greenplum — отказоустойчивая система. В ней есть primary-сегмент и зеркала. Если primary-сегмент упал, зеркало с ним синхронится. Оно поднимается, клиент переподключается к базе, и все работает. Но у нас на Greenplum завязана вся отчетность и аналитика, а количество пользователей и важность платформы данных для компании в целом очень велики. И если в ЦОД попадет метеорит или случится что-то еще, нам будет очень грустно без данных.
Так как наше хранилище строится на 90% SAS, у нас есть свой планировщик, который запускает SAS-джобы. Нам нужен был пообъектный перенос из этого планировщика. Представим простую ситуацию: на дворе ночь и мы строим Хранилище. У нас отработали 2000 джобов и осталось еще 2000 джобов впереди. И тут оно: в ЦОДе что-то пошло не так и мы его потеряли. И вместо того, чтобы перезапустить все, хочется перезапустить только половину, то есть продолжить построение с момента аварии.
А еще очень хочется иметь резервные копии всего, что мы уже построили: мало ли что еще может пойти не так. Обе эти проблемы отлично решает наш DUET. Да, сначала его называли Dual ETL, хотя по факту никакого двойного ETL/ELT не происходит.
В целом наши требования к disaster recovery для Greenplum были такими:
— запуск переноса из SAS-планировщика;
— создание и проверка бэкапов в ходе переноса;
— минимальная задержка доставки данных на резервный кластер;
— возможность регулировать нагрузку (количество потоков) на каждой стадии: бэкап, перенос, рестор, сброс на СХД;
— возможность использовать бэкапы для доставки данных на контуры тестирования и разработки.
Поначалу мы пробовали очевидные подходы: gp_transfer, gpfdist, gpfdist + pipes. Все работало замечательно, но бэкапов не было. Поэтому мы просто взяли большую NFS-шару и сделали на нее dump, а с нее — restore. Бэкапы появились, но мы быстро уперлись в производительность NFS и поняли, что с ростом нагрузки использовать ее будет все тяжелее и дороже.
В итоге мы пришли к следующей схеме:
— делаем dump на локальные диски (независимые от дисков БД, это были небольшие SAS SSD);
— на каждом сервере мы запускаем SCP, которая просто переносит это на соответствующие машины в другой кластер;
— на втором кластере запускаем рестор фактически из локальных файлов;
— если рестор прошел корректно — складываем все на хранилку (опять же NFS).
DUET первой версии
Стоит отдельно отметить, что хранилок было две (на двух площадках) и синхронизировались они самописным python-приложением. В какой-то момент это самописное приложение перестало вывозить наши потоки данных, и в итоге мы перешли на SDS LizardFS.
Показатели DUET на 2018 год:
— перегон 30 ТБ данных в сутки;
— задержка от построения объекта на боевой базе до доступности данных объекта на резервной базе данных — до двух часов;
— поддержка базы данных, частично работающей на зеркалах;
— постановка объектов в очередь SAS-планировщика;
— управление через веб-приложение.
При этом у первой версии DUET было несколько проблем:
— перенос управлялся python-скриптами, не было нормального API;
— локальные диски для бэкапов — фактически две записи одних и тех же данных (сначала бэкап, потом вынос в LizardFS отдельно);
— нет инкремента, что при нашем (иногда взрывном) росте — больно;
— и да, тогда мы переносили объекты полностью, даже не по партициям. А это снова боль от объемов;
— использование python-утилит из поставки Greenplum для бэкапа/рестора данных. Они создавали транзакции уровня serializable, из-за которых во всей БД не работал vacuum. Так и хомячка на куски разорвать недолго.
DUET2
Вторая версия нашей DR-системы привнесла следующие изменения:
— появление полноценного RestAPI;
— при этом для каждого кластера Greenplum мы сделали N очередей под разные задачи как на бэкап, так и на рестор;
— бэкапы мы стали сразу писать на LizardFS;
— вместо python-утилит мы перешли на использование COPY ON SEGMENT, что позволило уйти от serializable-транзакций;
— часть объектов переносилась по партициям.
Эта версия DUET переваривала уже до 70 ТБ в сутки. А кластеров Greenplum у нас на тот момент стало не три, а четыре.
Где-то во времена перехода к DUET3 кластеров Greenplum у нас уже стало восемь. Соответственно, реализация инкрементного переноса стала мегакритичной.
Вот как она работает сейчас:
Сперва ETL-планировщик при внесении изменений в таблицу использует insert-only diff-таблицу. Он записывает в нее строки, которые должны измениться в таргет-таблице, а потом из таргет-таблицы делает delete по ключам и insert данных. При этом он заполняет поле processed_dttm в основной и diff-таблице для новых/изменяемых записей одним значением. Эти diff-таблицы партицированы по дням и хранят партиции только за последнюю неделю. Поэтому они маленькие и запросы к ним работают быстро.
После отработки джоба ETL-планировщик откидывает в специальный сервис событие, в котором есть информация о произведенных изменениях: имя таблицы, processed_dttm, параметры diff (например, имя diff-таблицы, признак ее валидности, ключ для применения изменений, дополнительный фильтр для построения более оптимального запроса для delete-данных).
Событие обрабатывается и создается таск для DUET.
DUET по таску определяет, необходим ли полный перенос (full-таск) или достаточно перенести и накатить инкремент (diff-таск).
Если достаточно инкремента, делается дамп таблицы (select * from diff_table where processed_dttm = <>). Причем дамп только нужных записей для обновления таргет-таблицы.
Этот дамп сохраняется в LizardFS, а затем восстанавливается в соответствующие diff-таблицы на кластерах приемниках. Затем воспроизводится логика delete+insert нужных данных в основную таблицу из ETL-джоба.
Если в какой-то момент оказывается, что применение diff-таблицы невозможно, например на таргет-контуре у основной таблицы некорректный DDL или он вообще отсутствует, DUET сам себе ставит дополнительные таски на дамп и рестор полной таблицы.
DUET хранит историю дампов в виде цепочек. Цепочка имеет вид full [ + diff, diff, ...] — содержит полный дамп и некоторое количество diff-дампов. Соответственно, при необходимости можно восстановить любую версию таблицы из этой цепочки путем восстановления full и нужного количества diff.
DUET следит за тем, чтобы такие цепочки не были слишком большими. Периодически он сам себе ставит себе таски на снятие полного дампа в период минимальной загрузки, чтобы начать новую цепочку. Сейчас DUET успешно переваривает около 200 ТБ данных каждый день. При этом он доставляет данные построенной модели на семь кластеров Greenplum и обеспечивает задержку от построения на ETL-кластере до доступности данных пользователям до 30 минут.
Мы используем Greenplum почти десять лет. За это время мы хорошо изучили инструмент и поняли, как работать с ним эффективно. Вот, что мы делаем, чтобы упростить жизнь себе и пользователям.
Помним про shared-nothing. Greenplum — это MPP shared-nothing система. Она хорошо параллелится и масштабируется. Но MPP shared-nothing означает, что каждый маленький кусочек базы обрабатывает свой маленький кусочек данных. Следовательно, нужно стараться равномерно раскладывать данные по кластеру.
В Greenplum есть distribution key — ключ к распределению. Таблица может быть распределена рандомно по хэшу от всей строки или по одному или нескольким полям. Старайтесь раскладывать так, чтобы у вас все лежало ровно.
Архитектурно у всех подобных систем есть общая проблема: они работают со скоростью самого медленного сегмента. Если рейд просел по производительности на одном сервере, мы будем работать со скоростью этого рейда. Если мы положили данные криво и обращаемся к таблице, в которой они лежат, мы будем работать со скоростью этого запроса.
Кроме того, нужно учитывать нагрузку. Крайне желательно равномерно распределять временные данные в джойнах. Если запрос не вошел в память, он начинает писать диск. Соответственно, мы пишем кучу временных файлов на одну машину, а она тянет за собой кластер. Он не упадет, но будет работать медленнее.
Избегаем большого количества null-ов в ключах джойна. Об этом нюансе мы узнали, когда стали позволять большому количеству пользователей писать произвольный SQL в базе.
График ниже — это load average на кластере. Вы можете видеть, что один сервер выбивается. Больше того: с точки зрения базы выбивается один сегмент, на котором скапливаются эти null-ы. Сделать с этим в общем случае нельзя ничего. Можно только посоветовать пользователям отделять null-ы, чтобы потом их приджойнить.
Перекос нагрузки из-за null в ключах джойна
Режем количество и объем spill-файлов. Spill-файлы (спиллы) — это временные файлы, которые Greenplum складывает на диске, если запрос не вошел в память. Есть несколько параметров базы, которые влияют на спиллы. Это ограничение объема спиллов на запрос, ограничение количества спиллов на запрос и ограничение объема спиллов на сегмент.
Все началось с того, что база внезапно начала падать, когда в нее приходили запросы со множеством временных данных. Мы решили ограничить объем. Падать стали в два раза реже, но все равно падали, а почему — непонятно.
Поразмыслив, мы почитали документацию и сделали XFS под Greenplum с размером блока 16 МБ. Все заработало, но база писала очень много мелких файлов. Мы писали N сотен файлов по 4 КБ, и база видела их как файлы по 4 КБ, а файловая система и операционка с нами не соглашались, ведь у них размер блока — 16 МБ. Написали тысячу файлов — все забили на ровном месте. Тогда мы начали резать количество спиллов — и все стало хорошо.
Дальше мы поняли, что объем спилл-файлов — один из параметров, по которым можно и нужно убивать любые запросы к Greenplum вне зависимости от их важности. Да, «мне надо срочно выгрузить данные для ЦБ» — недостаточно хороший аргумент для того, чтобы писать плохие запросы.
Не злоупотребляем синхронизацией метаданных. У нас довольно большие кластеры: несколько десятков тысяч объектов и больше двух миллионов атрибутов. А клиенты, которые изначально не точились под Greenplum, любят синхронизировать метаданные маленькими кусочками, по чуть-чуть. При этом пользователей у наших кластеров много.
Это приводило к тому, что на аналитической базе было по 300—500 тысяч запросов к каталожным таблицам. Маленьких, но идущих постоянно. Greenplum это не любит из-за лишней нагрузки. Обнаружив это, мы попросили пользователей отключить автосинхронизацию меты. Казалось бы — мелочь, но вместо 300 тысяч запросов в сутки у нас теперь 0,5 тысячи, а это существенное сокращение.
Обучаем пользователей. Многие очевидные для нас вещи могут не быть очевидными для пользователей. Для них все выглядит как очень большой Postgres. И эта простота очень обманчива!
Однажды мы заметили, что у нас две ночи подряд висит сессия от пользователя в состоянии . Мы стараемся такого избегать: это либо очень большая загрузка, либо очень большая выгрузка, либо незакрытая транзакция, что в любом Postgres плохо. В нашем случае это приводило к замедлению базы, поэтому сессию пришлось оба раза убить.
Мы нашли пользователя и попытались выяснить у него, что он делал. Оказалось, он забирал табличку весом 60 ГБ в один поток через мастер. А мог в 140+ потоков через gpfdist, но он не умел этим пользоваться. Мы все ему объяснили, и выгрузка начала занимать у него 40 минут вместо 4 часов, не замедляя базу. И волки сыты, и овцы целы.
Greenplum имеет свою специфику. У нас на wiki есть учебник с лучшими и худшими практиками, который мы дополняем. Раньше мы рассылали худшие запросы в виде дайджеста. Берем самые долго висящие запросы от пользователей в базе, оптимизируем и рассылаем на всех пользователей хранилища. Объясняем, что было не так и как мы это исправили. Работает очень классно, всем советую.
Но сейчас мы все же перешли к хорошему, как нам кажется, онбордингу коллег. Сделали им несколько крутейших учебных курсов на внутреннем портале и даже команды Data Partners внутри нашего управления, которые помогают пользователям получать крутые результаты в работе с данными, никому при этом не мешая.
Greenplum, как и любая другая технология, — не серебряная пуля. Но он очень хорошо может решать определенные задачи на достаточно больших (up to petabyte scale) объемах данных и не самых маленьких нагрузках. Его близость к Postgres и открытый код дают простор для реализации различного рода идей и исправления недостатков системы. Ну и сама по себе открытость кода — очень большой бонус в нынешних условиях.
Но в эксплуатации важно понимать ограничения системы и правильно доносить их до пользователей. Либо закрывать систему от пользователей настолько, чтобы они не могли ее сломать, при этом получая требуемый уровень сервиса.
Наша платформа данных — одна из самых больших в России по количеству пользователей и нагрузкам, и Greenplum остается ее ядром. За его долгую жизнь мы вместе прошли через множество вызовов, многому научились сами и обучили наших пользователей, построили большое количество автоматизаций и продолжаем развивать нашу платформу. Если у вас остались вопросы, пишите комментарии.
