#debezium

Мы уже писали про поиск сложных событий при их потоковой обработке средствами Apache Flink. Продолжая эту важную для обучения дата-инженеров тему, сегодня рассмотрим, как CDC-коннектор от GetIndata упрощает запуск распознавание шаблонов на потоках данных из многих источников. Проблемы захвата измененных данных из реляционной базы с помощью JDBC-драйвера и способы их решения с Flink SQL.CEP и CDC в реальном времениВозможность в реальном времени обнаруживать шаблоны в потоках событий, чтобы заблаговременно принять соответствующие меры успешно решается в современных Big Data системах с помощью специализированных CEP-библиотек (Complex Event Processing, CEP). Они обеспечивают сопоставление потока входящих событий с заранее определенным шаблоном, на лету выполняя поиск релевантных данные по сохраненному запросу и отбрасывая неподходящие компоненты потока. CEP позволяет найти объект по шаблонизированному запросу в бесконечном потоке входных данных. Таким объектом может быть все что угодно, что имеет определенное поведение в виде последовательности событий. А шаблоном будет набор действий, которые может выполнять этот объект. Как только CEP обнаружит данные, удовлетворяющие заданному шаблону, система идентифицирует искомый объект в режиме реального времени.В Apache Flink идея CEP отлично реализуется благодаря потоковой природе фреймворка, а также его высокой пропускной способности обработки данных. Впервые API библиотеки CEP в Apache Flink был представлен в 2016 году. Библиотека FlinkCEP не является частью бинарного дистрибутива фреймворка, поэтому для ее использования следует явно задать зависимость FlinkCEP в файл pom.xml Flink-проекта, как мы показывали здесь.Идеи CEP коррелируют с захватом измененных данных (Change Data Capture, CDC) – подход, который работает с обнаружением изменения данных в исходной системе, чтобы взять именно эти изменения для дальнейшей обработки. Например, можно использовать CDC для сбора изменений данных в реляционной базе данных и создания потока событий, которые описывают эти изменения. Поток состоит из последовательности событий, которые описывают операции вставки, обновления и удаления строк таблицы. Разумеется, поток CDC может быть обработан Apache Flink, что позволяет выполнять сложные задания аналитики больших данных, в т.ч. распознавание паттернов.Чтобы посмотреть ,как это работает, разберем практический пример. Для большей наглядности сформулируем требование в виде пользовательской истории: «Как бизнес-аналитик, я хочу проверить эффективность маркетинговой кампании по быстрым кредитам с использованием мобильного приложения». Определим критерии приемки этой User Story как результат, который отображает список пользователей, которые:- выполнили транзакцию на сумму более 1000, используя мобильное приложение в течение 3 дней после транзакции;- не взяли кредит;- повторно использовали мобильное приложение в течение 7 дней после предыдущего случая.Предположим, исходные данные для этого кейса хранятся в нескольких источниках: в топиках Apache Kafka есть события транзакций и мобильных приложений, а PostgreSQL содержит информацию о пользовательских кредитах.Оперативное обновление результата в режиме реального времени обеспечат технологии потоковой обработки событий, в частности, Apache Flink, его CEP-библиотека для распознавания шаблонов, а также коннекторы. Как это реализовать, рассмотрим далее.CDC и коннекторы Apache FlinkApache Flink имеет несколько предустановленных коннекторов, например, к Kafka, который позволяет рассматривать топик как таблицу в Flink SQL. Благодаря этому можно обрабатывать информацию о транзакциях и событиях мобильных приложений. Однако, захват изменений от БД является более сложным сценарием: нужно преобразовать изменения данных из реляционных баз в качестве потока событий. Сделать это можно, используя готовые инструменты CDC. Одним из них является Debezium – популярный фреймворк с открытым исходным кодом, который поддерживается Red Hat. Он захватывает изменения данных из журналов транзакций БД и публикует соответствующие события в топиках Kafka, не снижая производительности источника. Debezium может собирать измененные данные в реальном времени, включая операции удаления и никогда не пропускает события, отражая все изменения в таблице.Недостатком Debezium является его сложная настройка, которая требует использования Apache Kafka, потребность в доступе к базе данных Binlog и ограниченный набор поддерживаемых баз данных.Альтернативой Debezium могут стать CDC-коннекторы от Ververica, которая занимается коммерциализацией и развитием Apache Flink. Flink-CDC-коннектор включает Debezium как CDC-движок, поэтому для него тоже характерны ранее указанные ограничения. Кроме того, этот коннектор поддерживает только MySQL (5.7, 8.0.x), PostgreSQL (9,6, 10, 11, 12) и MongoDB (4,0, 4,2, 5,0).Также для реализации CDC можно использовать официальный Flink Connector JDBC, который позволяет писать и читать данные из реляционных баз (MySQL, PostgreSQL, Derby) непосредственно во Flink SQL. Это встроенный инструмент фреймворка, который считывает данные из таблицы только один раз, поэтому его нельзя назвать средством полноценной поддержки CDC.Поэтому дата-инженеры компании GetIndata разработали собственный CDC-коннектор JDBC, который позволяет периодически читать данные из таблиц в реляционных базах. В отличие от Debezium, он не требует дополнительных компонентов и совместим с любой базой данных. Достаточно просто составить SQL-запросы для JDBC-драйвера. Коннектор использует чистый SQL для CDC, не нуждаясь в специальной конфигурации базы данных. Однако, этот коннектор не поддерживает операции удаления, а некоторые CDC-стратегии требуют дополнительных столбцов, например, last_modify_date. Работа коннектора может оказать влияние на базу данных, поскольку он считывает данные с помощью периодического запуска SQL-запроса, что потенциально снижает производительность. Коннектор GetIndata обновляет данные почти в реальном времени: логика CDC работает через определенные интервалы, которые можно настроить очень часто. Тем не менее, есть риск пропустить событие CDC в часто изменяющихся строках, например, когда вставка и удаление строки произошли раньше, чем коннектор обновил данные.Для разработки этого CDC-коннектора дата-инженеры GetIndata использовали табличный API, предоставленный Flink. Flink предоставляет интерфейсы, которые должны быть реализованы пользовательской логикой для обработки внешних источников данных, таких как таблица. Затем эту таблицу можно обработать с помощью Flink SQL. При этом выполнении запроса Flink не изменит никаких внешних данных: движок выполнения фреймворка использует определение таблицы для чтения всех данных из источника.Рассмотрим, как соединить идеи CDC с CEP, т.е. как определить распознавание шаблонов событий с помощью Flink SQL:SELECT *FROM eventsMATCH_RECOGNIZE(PARTITION BY customer_idORDER BY tsMEASURESTRX.event_id AS trx_event_id,TRX.customer_id AS trx_customer_id,TRX.type AS trx_type,TRX.payload AS trx_payload,TRX.ts AS trx_ts,APP_1.event_id AS app_1_event_id,APP_1.customer_id AS app_1_customer_id,APP_1.type AS app_1_type,APP_1.payload AS app_1_payload,APP_1.ts AS app_1_tsONE ROW PER MATCHPATTERN (TRX APP_1 NOT_LOAN*? APP_2) WITHIN INTERVAL '10' DAYDEFINETRX AS TRX.type = 'trx_event' AND TRX.payload > 1000,APP_1 AS APP_1.type = 'clickstream_event' AND APP_1.ts < TRX.ts + INTERVAL '3' DAY,APP_2 AS APP_2.type = 'clickstream_event' AND APP_2.ts > APP_1.ts AND APP_2.ts < APP_1.ts + INTERVAL '7' DAY,NO_LOAN AS NOT_LOAN.type 'loan_event') MR;В вышеприведенном коде определены события из ранее описанной пользовательской истории о проверке эффективности маркетинговой кампании по быстрым кредитам с использованием мобильного приложения:- TRX - событие, которое описывает транзакцию с суммой более 1000;- APP_1 - событие, представляющее взаимодействие с пользователем с приложением в течение 3 дней после транзакции;- NO_LOAN - событие, которое сообщает, что пользователь не взял кредит.Также в коде указан порядок событий и ожидаемый выход, чтобы найти шаблон из этой пользовательской истории. Для определения шаблона используется регулярное выражение.Освойте тонкости использования Apache Flink и Spark для потоковой обработки событий в распределенных приложениях аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:- Потоковая обработка данных с помощью Apache Flink- Потоковая обработка в Apache SparkИсточники1. - https://getindata.com/blog/change-data-capture-JDBC-Flink-FlinkSQL/2. - https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/libs/cep/

Недавно мы писали про Lakesoul – новое унифицированное решение для хранения потоковых и пакетных таблиц, которое реализует архитектуру данных LakeHouse. Сегодня заглянем под капот этого унифицированного механизма на базе Apache Spark и разберемся с преимуществами его последнего релиза.Как работает LakeSoul: краткий обзорНапомним, LakeSoul от команды DMetaSoul представляет собой унифицированный механизм хранения потоковых и пакетных таблиц на базе Apache Spark, который поддерживает масштабируемое управление метаданными, ACID-транзакции, эффективную и гибкую операцию вставки-изменения (UPSERT) данных, эволюцию схемы, а также потоковую и пакетную унификацию. Это решение соответствует новой гибридной архитектуре данных под названием LakeHouse, которая стремится объединить достоинства классических DWH с преимуществами гибкого озера данных (Data Lake), о чем мы писали здесь и здесь.LakeSoul поддерживает многоуровневое управление разделами, несколько разделов диапазона и один хеш-раздел на уровне раздела таблицы. В реальном бизнес-сценарии большое хранилище данных будет отправлять большой объем информации о разделах на уровень метаданных после длительного обновления. Для сценариев обновления в реальном времени отправка происходит чаще. При этом часто встречается инфляция метаданных, что приводит к низкой эффективности доступа к метаданным. Производительность метаданных сильно влияет на производительность запросов, поскольку во время запроса данных необходимо получить доступ к информации о разделах и другой базовой информации о распределении данных в метаданных. Поэтому для хранилища данных важна высокопроизводительная и расширяемая служба метаданных.LakeSoul использует мощную распределенную NoSQL-СУБД Cassandra для управления метаданными. Это децентрализованное масштабируемое решение предоставляет богатые методы моделирования данных и высокую пропускную способность чтения/записи. В качестве хранилища метаданных Cassandra обеспечивает настраиваемые уровни доступности и согласованности, простое управление параллелизмом и ACID для операций с метаданными.LakeSoul организует первичный ключ и индекс таблицы уровня метаданных. Для раздела конечного уровня требуется только одна операция с первичным ключом, чтобы получить всю информацию об этом разделе, прочитать и записать моментальный снимок (snapshot) текущей версии и пр. Snapshot раздела содержит полный путь к файлу и тип фиксации для полной записи и операций upsert. Этот план чтения раздела может быть построен путем последовательного обхода фиксаций файлов в моментальном снимке. Такой доступ к информации о разделах позволяет избежать обхода файлов и каталогов, необходимого метода оптимизации для объектных хранилищ типа AWS S3 и OSS.В отличие от Apache Hive, которая использует слабомасштабируемую реляционную СУБД MySQL в качестве уровня хранения метаданных, благодаря Cassandra LakeSoul поддерживает на порядок больше разделов в одной таблице. Аналогичное преимущество LakeSoul можно отметить по сравнению с Apache Iceberg и Delta Lake, которые используют файловые системы для хранения метаданных.Чтобы обеспечить согласованность одновременной записи и чтения данных, LakeSoul поддерживает транзакции ACID и одновременные обновления. В отличие от OLTP-сценария, обновления Lakehouse являются гранулированными на уровне файлов. LakeSoul использует облегченную транзакцию Cassandra для реализации обновлений на уровне разделов. Как и многие NoSQL-СУБД, Cassandra по-своему реализует ACID-требования к транзакциям. В частности, она может обеспечить обновленную атомарность и изоляцию с помощью LWT (Light Weight Transaction) – механизма легковесных транзакций. А доступность и согласованность можно контролировать с помощью гибко настраиваемых уровней согласованности Cassandra.В частности, когда вычислительный механизм создает файлы для каждого раздела для фиксации, он сначала фиксирует информацию об обновлении файла раздела, такую ​​как полное или добавочное обновление (upsert), а затем обновляет видимую для читателя версию через LWT. В сценариях, где обнаруживаются одновременные обновления, LakeSoul автоматически различает типы записи, чтобы определить, есть ли конфликт, и следует ли повторно отправить данные напрямую или откатить операцию.LakeSoul поддерживает несколько механизмов обновления, включая Append, Overwrite и Upsert. Общие записи потоковой передачи журнала обычно находятся в Append. В этом случае уровень метаданных должен записывать только пути к файлам, добавляемые каждым разделом. В то же время задание на чтение данных единообразно считывает все присоединенные файлы в разделе, чтобы выполнить слияние при чтении. Для ситуаций перезаписи, когда обновление/удаление происходит в произвольных условиях или когда происходит уплотнение, LakeSoul использует механизм копирования при записи для обновления.LakeSoul поддерживает более эффективный механизм Upsert в случае хеш-разделов и операций для хеш-ключей. В каждом сегменте хеширования LakeSoul сортирует файлы по ключу хеша. После многократного выполнения Upsert получается несколько упорядоченных файлов, которые сливаются при операции чтении.Операции Upsert и MergeOnRead в LakeSoulТаким образом, LakeSoul достигает высокой пропускной способности записи при сохранении скорости чтения за счет оптимизированного упорядоченного слияния файлов. Чтобы предоставить эффективные возможности хранилища данных в реальном времени, LakeSoul поддерживает сценарии захвата обновленных данных (CDC, Change Data Capture), предоставляя отдельный набор семантических выражений. LakeSoul может взаимодействовать с различными источниками сбора измененных данных, такими как Debezium, Flink и Canal, путем указания столбца операции в таблице CDC. Этот столбец действий источника коллекции можно преобразовать в столбец CDC-действий LakeSoul. В случае онлайн-синхронизации с OLTP-СУБД принимающая таблица LakeSoul также может использовать метод хэширования и записывать CDC в Upsert для хешированных ключей, достигая очень высокой скорости.На практике это полезно в таких сценариях, как синхронизация CDC-отчетов в режиме реального времени с хранилищами и витринами данных. Если для этого использовать базу данных периодического дампа, задержки и накладные расходы на хранение данных становятся слишком велики. А движки для вычислений в реальном времени типа Apache Flink требуют много ресурсов для разработки и обслуживания приложений. CDC-синхронизация с LakeSoul в реальном времени и преобразование в Upsert в случае операции с первичным ключом позволяют достичь чрезвычайно высокой пропускной способности записи. Изменения в базе данных могут быть синхронизированы с DWH практически в режиме реального времени, чтобы быстро выполнить BI-анализ онлайн-данных с помощью SQL-запроса. С этой задачей отлично справляется Debezium - сервис захвата изменений в базах данных и отправки их на обработку в другие системы, о чем мы писали здесь и здесь. Применение CDC в Data Lake может значительно упростить архитектуру обновления данных в режиме реального времени.Также отметим еще один пример практического использования LakeSoul для построения образца базы данных рекомендательной системы в режиме реального времени. Общим требованием в этом сценарии является объединение нескольких таблиц, включая характеристики пользователей, характеристики продукта, метки воздействия и метки кликов, для библиотеки обучающих примеров ML-модели. Офлайн-соединение больших таблиц имеет проблемы с низкой своевременностью и высоким потреблением вычислительных ресурсов. Обойти это ограничение можно, используя Flink для выполнения многопотокового соединения в реальном времени. Но в случае большого временного окна это потребляет много ресурсов. LakeSoul может преобразовать многопотоковое соединения в параллельный Upsert. Поскольку разные потоки имеют один и тот же первичный ключ, его можно установить на ключ хеш-раздела для Upsert, что даже в случае большого временного окна обеспечит высокую пропускную способность записи.Что нового во 2-ой версии и при чем здесь Apache SparkВ конце июля 2022 года команда DMetaSoul выпустила новую версию LakeSoul - 2.0, где служба каталога отделена от Apache Spark, чтобы реализовать независимую структуру и интерфейс хранения метаданных. Spark, Flink, Presto и другие вычислительные движки могут взаимодействовать с LakeSoul для обеспечения потоковой передачи и пакетной интеграции. Теперь DDL-запросы и функции датасета в Spark SQL тесно взаимодействуют с каталогом. Информация о созданной таблице хранится в Каталоге. Для этого в LakeSoul перенастроен интерфейс Spark Scala.Приемник данных (DataSink) Spark включает в себя не только информацию о метаданных, но и обнаружение конфликтов данных файлов данных и разделов, когда потоковая пакетная задача записывает данные. Поэтому скорректирован интерфейс работы с данными после завершения задания Spark. Обнаружение конфликтов фиксации данных секционирования также перенесено в Каталог.Слияние при чтении (Merge On Read) оптимизировано для источника данных Spark. Ранее при чтении данных файлы сортировались по номеру версии записи. Поэтому при слиянии последняя версия по умолчанию перезаписывала данные прежней версии. В LakeSoul 2.0 атрибут записи версии для файлов данных удален вместо использования упорядоченного списка файлов с последним файлом сборки в конце списка.В заключение отметим, что LakeSoul 2.0 также поддерживает унифицированную потоковую передачу и пакетную запись Flink, а также реализует семантику строго однократной доставки сообщений exactly-once. Это обеспечивает точность данных для Flink CDC и других сценариев, сочетая возможности Merge On Read и Schema AutoMerge, что важно для слияния нескольких таблиц и баз данных.Узнайте больше про архитектуру и аналитику больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:Архитектура ДанныхПотоковая обработка в Apache SparkИсточники1. https://dev.to/qazmkop/the-best-open-source-lakehouse-project-lakesoul-20-supports-snapshot-rollback- flink-and-hive-interconnection-49ni2. https://www.libhunt.com/compare-MetaSpore-vs-LakeSoul3. https://github.com/meta-soul/LakeSoul/issues/23