Часто при построении высоконагруженных проектов задействуют несколько хранилищ. При этом нередко одновременно используют как SQL-, так и NoSQL-базы данных. Такая реализация оправдана, но для её корректной работы надо правильно настроить репликацию данных в одну из сторон.
Меня зовут Александр Горякин, я разработчик высоконагруженных систем хранения данных в пресейле Tarantool. Не так давно наша команда столкнулась с необходимостью репликации между SQL- и NoSQL-базами данных и успешно решила эту задачу. Я расскажу, что нам нужно было от репликации, какие механизмы и инструменты мы рассматривали, с какими проблемами столкнулись и что получили в итоге.
Статья написана по моему выступлению на HighLoad++ 2022. Вы можете посмотреть его здесь.
Задачи и исходные требования
Перед нами стояла задача переливки данных между разными хранилищами в разных конфигурациях — как SQL, так и NoSQL. При этом у нас были высокие требования к репликации. В том числе:
- Низкий лаг. Было важно, чтобы между попаданием записи в источник и в приёмник было не больше 40-50 мс.
- Низкая нагрузка на БД-источник. Большинство наших клиентов Enterprise-уровня. Они используют высоконагруженные системы, и дополнительная нагрузка на источник часто недопустима с точки зрения отказоустойчивости.
- Устойчивость к высоким нагрузкам. Даже при пике нагрузки важно, чтобы лаг репликации был постоянным или имел минимальные отклонения.
- Отказоустойчивость при сбоях. В случае падения сети или перезапуска виртуальной машины с продуктом, после устранения сбоя репликация должна продолжить выполняться в штатном режиме.
Варианты интеграции баз данных: ETL, CT, CDC
Есть несколько основных вариантов интеграции баз данных:
- ETL (Extract Transform Load).
- CT (Change Tracking).
- CDC (Change Data Capture).
Extract Transform Load
ETL — универсальное решение, которое можно реализовать на любой базе данных. При этой модели репликации вся информация из БД-источника с помощью стека инструментов выгружается в человекочитаемый формат (CSV или JSON), а потом с помощью другого стека загружается в БД-приемник.
У такого механизма два преимущества:
- Универсальность. ETL позволяет выбирать все записи в любом хранилище.
- Удобство. С выгружаемыми из источника данными легко работать.
Но есть и недостатки:
- Дополнительная нагрузка. При ETL проводится полное сканирование источника, поэтому нагрузка на систему кратно увеличивается.
- Низкая скорость. Из-за нескольких промежуточных звеньев и операций у репликации через ETL огромный временной лаг.
- Нужно много места. Для хранения данных, которые выгружаются из источника в промежуточном формате, нужен большой объём диска. Чем больше, тем сложнее обеспечить инфраструктуру.
Учитывая недостатки, ETL для высоконагруженных систем не подходит.
Change Tracking
CT — метод, при котором изменения в источнике попадают в журнальную таблицу, которую вычитывают инструментом и, в случае обнаружения изменений, передают данные в БД-приемник.
Плюс такого механизма — в стриминге из источника: при попадании записи об изменениях в журнальную таблицу данные сразу реплицируются. Как результат, скорость высокая, а лаг — низкий.
Но есть и недостатки:
- Изменения приходят в виде дельты. Нельзя получить данные до и после изменения, только изменённые фрагменты.
- Ограниченная поддержка. Не все базы данных «из коробки» поддерживают репликацию через Change Tracking.
- Дополнительная нагрузка на источник и приёмник. Из-за выгрузки из промежуточной журнальной таблицы нагрузка на узлы системы увеличивается.
- Отсутствие общих гарантий. Гарантии консистентности и инструментарий уникальны для каждого отдельного кейса. Это связано с тем, что источники, промежуточные таблицы и приёмники в разных проектах различаются.
Под наши задачи механизм Change Tracking подходил только частично — нам было важно реализовать репликацию без дополнительной нагрузки на источник и приёмник. С Change Tracking сделать это нельзя.
Change Data Capture
Механизм CDC подразумевает, что все изменения в базе данных фиксируются в журнале опережающей записи (WAL). С помощью стека инструментов записи из журнала читают и загружают в приёмник. Причём читать журнал опережающей записи можно напрямую или по репликационному протоколу.
У Change Data Capture есть несколько ключевых преимуществ:
- Изменения приходят полностью. Это удобно — известно, как выглядела запись до изменений и какой вид имеет сейчас.
- Работа через WAL. Чтение записей выполняется через WAL, который выступает промежуточным звеном. Благодаря этому, нагрузка на источник не увеличивается, достигается высокая скорость репликации и низкий лаг.
- Гарантированная консистентность данных. Согласованность достигается за счет использования репликационных протоколов баз данных.
Из недостатков Change Data Capture только рост нагрузки на БД-приёмник. Но это неизбежный «побочный эффект» — загрузить данные в приёмник, не увеличивая количество запросов, просто невозможно.
В итоге мы остановились на механизме CDC.
Решения на основе CDC
Change Data Capture — принцип репликации, а не сам инструмент. Его нам предстояло выбрать. При этом надо было учесть, что у большинства наших клиентов основу стека формирует Oracle и Java. Исходя из этого было несколько вариантов.
Oracle GoldenGate. Решение, с помощью которого можно извлекать изменения из redo-логов баз данных Oracle. Также в Oracle GoldenGate пишет библиотека userexit, которая позволяет «переливать» изменения в приемник.
Это быстрый инструмент, но он очень сложен в настройке и дорогой: цена лицензии превышает несколько десятков тысяч долларов. Кроме того, GoldenGate работает в основном с Oracle.
IBM Infosphere CDC. Аналог Oracle GoldenGate для для IBM DB2. Он быстрый, но сложный, дорогой и с ограниченной поддержкой: только для IBM DB2.
Самописный CDC-tool. Можно было самим разработать нужное решение — например, взять библиотеки pglogrepl и pgproto, которые позволяют парсить протоколы PostgreSQL и логической репликации, и на их основе реализовать приложение.
Самописная реализация хороша тем, что полностью понятен принцип работы инструмента и можно адаптировать его под конкретную задачу. Но для разработки полноценного решения нужно много времени и ресурсов, а «на выходе» можно получить узкоспециальный продукт, который сложно поддерживать.
Debezium. Сервис для захвата изменений в базах данных (Change Data Capture) и отправки их на обработку в другие системы. У Debezium есть Source-корректоры к БД-источникам — изначально они работают поверх Apache Kafka Connect, но с помощью библиотеки Debezium Embedded коннектор можно встроить в приложение без Kafka. Также есть Debezium Server, готовое приложение для организации обмена данными при их миграции или модернизации существующих систем. В свою очередь в Debezium Server доступны Sink-коннекторы, с помощью которых можно передавать информацию в целевые приёмники: Redis, Kafka, Apache Pulsar и другие. Более того, Debezium написан на Java, то есть он подходит под стек наших заказчиков.
Мы выбрали этот вариант.
Работа с Debezium: с чего мы начали
Изначально нам казалось, что сделать решение на Debezium можно быстро и легко — достаточно изучить документацию и получится заменить Oracle GoldenGate, который к тому моменту был у нас основным. Так мы приступили к разработке приложения Debezium Embedded. В процессе разработки мы обнаружили, что коннектор хорошо справляется с чтением redo-логов с помощью LogMiner и изменений с помощью XStream API, но использовать его очень сложно.
Причины крылись в задействованных инструментах:
- LogMiner нестабилен: в любой момент он может прислать сообщение, которое коннектор не может распарсить, что приведёт к сбою всего решения.
- LogMiner deprecated: в последних версиях Oracle LogMiner не умеет передавать данные потоково.
- Спецификация LogMiner меняется от версии к версии.
- XStream API работает по лицензии Oracle GoldenGate, переход на него не решает проблему высокой стоимости.
Это была наша первая ошибка. Но в результате мы получили прототип, хоть и не нашли для него применения.
От сложного к простому: доработка прототипа
Через полгода мы решили доработать прототип, но использовать более понятный и простой инструмент, вроде PostgreSQL, и обеспечить универсальность решения. На выходе мы получили монолитное приложение, которое содержит в себе несколько коннекторов и позволяет потоково передавать данные из PostgreSQL в Tarantool. Но даже после замены инструмента лаг репликации оказался высоким — до нескольких минут — и рос пропорционально количеству записей. Одновременно с этим мы узнали и о других проблемах, которые надо было решить.
Высокий лаг репликации
Выявленный в Debezium Embedded лаг решили уменьшить несколькими способами.
- Организовали асинхронную загрузку данных. Мы начали отправлять данные в приёмник и не дожидаясь ответа о применении изменений. Но при изучении логов Tarantool оказалось, что данные приходят без сохранения порядка отправки, в итоге данные в целевой БД получаются неконсистентными.
- Перешли от JSON к Kafka SourceRecord. На парсинг JSON надо много времени. Мы решили заменить его на Kafka SourceRecord — это бинарный, нативный формат Kafka, на парсинг которого надо в разы меньше времени. Изменение помогло, но только частично: лаг репликации уменьшился, но всё равно достигал нескольких минут.
- Применили батчинг. Данные посылались из того же потока, в котором приходили из приёмника, и изначально отправлялись по одной записи. Мы решили собирать данные в батч и в таком виде отправлять их в приёмник. Благодаря этому лаг репликации уменьшился в 1,5-2 раза, но всё равно оставался высоким.
- Разделили потоки данных. Мы разнесли поток приёма из источника и поток отправки в приёмник на разные операции. Между потоками добавили трансфер (очередь или канал), который позволял сформировать батч нужного размера. В результате разделения лаг репликации уменьшился до 40-50 мс.
Низкая отказоустойчивость
Следующая проблема, с которой мы столкнулись — отказоустойчивость. Debezium хранит оффсеты — отметки о прочтении записей — в файле, который легко удалить, или в Kafka, которая может быть недоступна. Если оффсеты будут недоступны, то решение начнёт повторно перечитывать все изменения из источника, что неизбежно увеличит нагрузку на всю систему.
Мы решили организовать хранение оффсетов в БД-приёмнике. В таком случае при отправке батча в Tarantool автоматически сохраняется оффсет для каждой записи, который при запуске можно прочитать. Для считывания оффсетов мы написали отдельный класс для Debezium, который читает оффсеты при старте и стартует с последнего полученного.
Также на отказоустойчивость изначально влияли ошибки записи в БД-приёмник. Так, если приёмник шардированный, то часть изменений может не применяться, даже если записи поступают исправно. Debezium Embedded «из коробки» не распознаёт эту проблему — он уведомляет, что изменение применено, если даже по факту это не так.
Чтобы устранить этот баг, мы настроили повторные попытки записи батча — если изменение не применено, LSN-записи возвращаются и батч отправляется повторно. Это позволяет гарантировать консистентность данных в БД-приёмнике.
Монолитная архитектура
Монолитная архитектура также стала для нас проблемой. Конечно, при такой реализации все коннекторы доступны «из коробки» и сам монолит легко настраивать: достаточно указать профиль конфигурации и описать подключение к источнику и приёмнику в конфигурационных файлах. Но в нашем случае монолитная архитектура не подходила по ряду причин:
- Сложность масштабирования. У нас два потока данных, и при такой архитектуре нельзя масштабировать приложение на несколько ReplicaSet — надо только запускать новые инстансы.
- Невозможность адаптации. В монолите любое изменение внутреннего компонента приводит к необходимости переработки всего приложения. Это сложно и дорого.
Для перехода с монолитной архитектуры мы решили использовать Debezium Server, который:
- содержит в себе source-коннекторы для подключения к БД-источникам;
- имеет sink-коннекторы для подключения к БД-приёмникам;
- реализует функциональность, которую мы сделали для Debezium Embedded.
Кроме архитектуры, Debezium Server привлёк нас и преимуществами:
- Есть готовые сборки — все дистрибутивы можно просто скачать с сайта.
- Легко добавить новые source-коннекторы, в том числе работать с собственными.
- Конфигурация очень простая: достаточно в конфигурационном файле указать БД-источник, БД-приёмник и параметры подключения.
- Инструмент позиционируют как решение для Kubernetes, поэтому его легко масштабировать с помощью k8s.
- Позволяет добавлять своё хранилище оффсетов, что мы и сделали в Debezium Embedded.
Но у него есть и недостатки:
- Сложность добавления своего sink-коннектора: надо указывать зависимости и вносить изменения в конфигурационный файл.
- При каждом изменении sink-коннектора надо пересобирать дистрибутив или коннектор.
- Из коробки оффсеты хранятся в файле, в Kafka или в Redis — как и в случае с прототипом это не лучший вариант.
В нашем случае преимущества оказались важнее недостатков, поэтому Debezium Server стал нашим основным решением.
Решения для работы с распределёнными источниками
БД-источник может быть шардированной, как, например, Tarantool. Каждый из инструментов работает с такими БД по-разному.
- Debezium Embedded. Масштабирование выполняется по шардам и вручную, оффсеты хранятся в приёмнике для каждого шарда.
- Debezium Server. Масштабирование выполняется по шардам с помощью Kubernetes. Оффсеты хранят в файле, Kafka, Redis, приемнике для каждого шарда.
- Kafka-connect. Масштабирование выполняется по шардам с помощью тасков Kafka-connect — для каждого ReplicaSet запускается отдельный таск, который будет считывать данные. Оффсеты для каждого шарда хранятся в Kafka.
Главное из нашего опыта
- Зачастую к репликации высокие требования. Нам были важны скорость, отказоустойчивость, отсутствие дополнительной нагрузки. Исходя из этих требований мы выбрали механизм Change Data Capture, который удовлетворял всем основным критериям.
- Инструмент надо выбирать с учётом стека конечных пользователей. Наши заказчики часто работают с Java, поэтому, чтобы обеспечить совместимость решения, в качестве основы стека мы выбрали Java и Debezium.
- Монолитные приложения не всегда хороши. Debezium Embedded было сложно масштабировать и конфигурировать, поэтому мы перешли на Debezium Server. Это универсальное не монолитное решение, которое позволяет добавлять свои коннекторы и масштабироваться с помощью Kubernetes.
- Ошибки при создании инструмента для репликации неизбежны. Мы столкнулись с рядом проблем: высокий лаг, низкая отказоустойчивость, сложная архитектура. Каждый из недостатков нам пришлось устранять отдельно. Но тщательная доработка помогла нам получить инструмент, с помощью которого можно переливать данные из PostgreSQL в Tarantool с низким лагом репликации и сохранением консистентности.
Скачать Tarantool можно на официальном сайте, а получить помощь — в Telegram-чате.
