Потоковая аналитика Big Data с Oracle, Kafka и CDC Source Connector

В феврале 2021 года разработчики корпоративной версии Apache Kafka с коммерческой поддержкой, компания Confluent, выпустили премиум-коннектор к Oracle – одной из главных реляционных баз данных мира enterprise. Разбираемся, кому и зачем это нужно, а также как устроена такая интеграция SQL-СУБД и потоковой аналитики Big Data с применением CDC-подхода.

Реляционный монолит Oracle в мире потоковых микросервисов и Big Data

Выпуск нового Kafka-коннектора для Oracle Database – это не просто попытка соединить проприетарную монолитную СУБД с open-source технологиями Big Data, а настоящая интеграция OLTP-систем с современной платформой потоковой аналитики больших данных. Oracle давно и прочно занимает лидирующее место в мире корпоративных СУБД, обеспечивая дорогостоящее, но надежное хранение и аналитику структурированных данных в банках, ритейле, телекоммуникационных компаниях, государственных учреждениях и многих других крупных организациях. Однако, сегодня бизнесу нужно анализировать не только структурированные данные в рамках Oracle, но и множество другой информации, в т.ч. разных форматов и в режиме реального времени. К примеру, требуется обрабатывать данные из внешних источников и/или управлять потоковыми приложениями. Эту проблему решает подход к отслеживанию измененных данных (CDC, Сhange Data Capture), позволяя быстро выявлять и собирать информацию, которая была добавлена, обновлена или удалена из таблиц в Oracle, чтобы сделать это доступными для других систем и подразделений компании.

Реализация собственного CDC-решения для Oracle с регистрацией важных событий изменения базы данных потребует слишком много ресурсов и в перспективе приведет к росту технического долга. А, с учетом тренда на event-streaming архитектуру и аналитику больших данных в реальном времени с Apache Kafka, о чем мы писали здесь, интеграция Oracle Database с этой платформой потоковой передачи событий становится особенно востребованной. Именно поэтому Confluent выпустил Oracle CDC Source Connector, который позволяет надежно и экономично синхронизовать данные в режиме real-time, выгружая информацию из Oracle Database в Apache Kafka. Благодаря этому можно реализовать множество современных Big Data кейсов, от обнаружения мошенничества в реальном времени до модернизация корпоративных хранилищ данных [1]. Как устроен сам коннектор и какие именно технические возможности он предоставляет разработчику Kafka-приложений и дата-инженеру, мы рассмотрим далее.

Как работает Kafka CDC Source Connector

Коннектор Confluent для Oracle CDC Source v1.0.0 использует Oracle LogMiner для чтения kjuf повторов базы данных и требует дополнительного логгирования всех столбцов для нужных таблиц СБУД или всей базы данных. Пока Kafka-коннектор поддерживает Oracle Database 11g, 12c и 18c, в ближайшем будущем ожидается поддержка 19c. Синхронизация данных запускается с моментального снимка (snapshot) таблиц, считывает логи с определенного номера системного изменения Oracle (SCN, system change number) или с временной метки (timestamp) [1].

Коннектор использует технологию Oracle Change Data Capture, передающую обновления, вставки и удаления из базы данных OLTP в специальные таблицы, которые можно опрашивать, не влияя на параллелизм в обновляемых первичных таблицах. Затем коннектор переносит эту информацию в Kafka по принципу один топик для каждой таблицы СУБД, откуда данные можно дальше распространять в целевые места назначения без дополнительной нагрузки на исходный сервер базы данных Oracle. Таким образом, Kafka выполняет роль сервисной шины предприятия (ESB, Enterprise Service Bus), обеспечивая микросервисный подход для интеграции корпоративных систем и данных [2]. Благодаря гибким возможностям настройки конфигурационных параметров этого Kafka-коннектора, можно создавать собственные CDC-конвейеры между Oracle Database и BigQuery, Snowflake, MongoDB Atlas, Amazon Redshift и множества других аналитических платформ [1].

Premium Connector для Oracle CDC позволяет командам разработчиков безопасно фиксировать изменения в СУБД Oracle, и сохранять их в виде различных топиков Kafka, обеспечивая следующие технические сценарии использования [3]:

логгирование повторов для защиты транзакционных данных в Oracle с помощью резервного копирования лога повторного выполнения в распределенную отказоустойчивую систему Kafka. Если журнал повторного выполнения Oracle поврежден или несовместим с текущей схемой таблицы, коннектор отправляет его блок в раздел журнала ошибок. Пока коннектор поддерживает запись только в топик журнала повторов с одним разделом. Все преобразованные журналы повторного выполнения отправляются в один и тот же раздел.
моментальные снимки для включения обработки последних событий изменений в Oracle без временной задержки и синхронизации пользовательских приложений с текущим состоянием каждой таблицы корпоративной СУБД;
топики событий изменения таблиц для предотвращения несанкционированного доступа к данным изменений Oracle за счет разделения событий изменения каждой таблицы на разные топики Kafka;
ключи записей для надежного обмена данными в Oracle с помощью последовательной обработки и легкой идентификации, какая запись была изменена первичным ключом;
автоматическая синхронизация набора таблиц и реконфигурация задач при создании и удалении таблиц Oracle прямо во время работы коннектора. Когда коннектор определяет новые или удаленные таблицы, он автоматически перенастраивает свои задачи, чтобы прекратить наблюдение за удаленными таблицами и начать сбор изменений из новых таблиц, которые соответствуют выражениям фильтров таблиц.
масштабирование рабочих нагрузок базы данных с помощью параметра настройки максимального числа задач tasks.max.
микроперебалансировка рабочих нагрузки (для версий Confluent Platform 6.0 и более), когда коннектор равномерно распределяет таблицы по своим задачам, отслеживает изменения пропускной способности для каждой таблицы и положение каждой задачи в журнале повторов. Коннектор автоматически пытается распределить нагрузку между всеми задачами коннектора, назначая часто изменяющиеся таблицы различным задачам.
автоматическое создание топиков Kafka (для версий Confluent Platform 6.0 и более). В конфигурацию коннектора можно включить правила определения параметров топика Kafka, в который выполняется запись.
автоматическое переподключение при обрыве соединения с базой данных. Когда соединение потеряно, коннектор останавливается, регистрирует предупреждение о разъединении или сообщения об ошибках и пытается повторно подключиться. Как только соединение будет восстановлено, соединитель автоматически возобновит нормальную работу. Это поведение контролируется несколькими свойствами соединения, включая timeout.ms (по умолчанию 5 минут) и max.retry.time.ms (по умолчанию 24 часа). Отменить автоматическое переподключение можно, установив конфигурационный параметр max.retry.time.ms равным 0.
поддержка мультиарендной архитектуры Oracle CDB/PDB в версиях Oracle Database 12c и выше. Системные таблицы хранятся в единой контейнерной базе данных (CDB, Container DataBase). Пользовательские таблицы хранятся в подключаемых базах данных (PDB? Pluggable DataBase), подключенных к CDB. Каждый экземпляр коннектора может читать пользовательские таблицы в одной PDB. Имя PDB, в котором находятся пользовательские таблицы, можно настроить с помощью свойства pdb.name. Чтобы читать из системных таблиц в CDB или из устаревшей версии 11g, следует оставить свойство конфигурации oracle.pdb.name пустым. Для свойства oracle.sid необходимо задать системный идентификатор (SID, system identifier) Oracle, чтобы получить доступ к CDB, PDB или устаревшей немультиарендной СУБД.
интеграция с защищенным протоколом Kerberos благодаря свойству конфигурации kerberos.cache.file, в котором указывается расположение файла кэша билетов.

Справедливости ради стоит отметить некоторые ограничения этого коннектора к Kafka:

поддерживаются не все типы данных: коннектор не различает числовые типы INT, INTEGER, SMALLINT, DEC, DECIMAL, NUMBER, NUMERIC, объединяя их в логический тип Connect Decimal. А числовые типы с плавающей запятой DOUBLE PRECISION, REAL и FLOAT преобразуются в Connect FLOAT64.
поддерживаются не все SQL-операторы. После запуска коннектор распознает и анализирует операторы DDL, примененные к базе данных. Эти операторы DDL используются для идентификации изменений в структуре захваченных таблиц и для настройки схемы записей событий, отправляемых в Kafka. DDL-анализатор коннектора не поддерживает SQL-операторы ALTER TABLE для добавления или удаления ограничений первичного ключа, а также столбцов и переименования таблиц или столбцов.
невозможность записи сводной информации о транзакциях;
не поддерживается Protobuf-конвертор;
особенности преобразований одного сообщения (single-message transformation, SMT), которые чувствительны к схеме записи.

Тем не менее, несмотря на указанные ограничения, Confluent Oracle CDC Source Connector, созданный разработчиками Apache Kafka, облегчает ежедневную работу дата-инженеров и разработчиков распределенных приложений потоковой аналитики Big Data, позволяя снизить затраты на интеграцию данных и сократить сроки вывода готовых продуктов на рынок [3].

Как работает потоковый CDC в Apache Kafka на примере платформ Connect и Debezium, мы рассматривали в этой статье. А на практике освоить все тонкости разработки и администрирования кластеров Apache Kafka вы сможете на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:

Смотреть расписание

Записаться на курс

Источники