XIX Международная студенческая научная конференция Студенческий научный форум - 2027

Введение

Синхронизация продуктивных и аналитических хранилищ требует своевременного применения изменений. Аналитическая система ориентирована на чтение и агрегацию больших объемов данных, поэтому частые обновления и удаления могут создавать существенные накладные расходы. Потоковая обработка и микропакеты позволяют объединять поступающие операции перед записью результата [1, 2]. При выборе архитектуры необходимо сопоставлять задержку, пропускную способность, восстановление после сбоев и сложность реализации. Сравнения систем обработки потоков показывают значимость условий испытаний [3, 4]; результаты разных конфигураций нельзя непосредственно переносить на конкретный конвейер синхронизации.

Цель исследования

Исследовать архитектуры применения потоковых CRUD-изменений к аналитическому хранилищу, спроектировать и реализовать два выбранных подхода и сравнить условия их практического использования. Рассматриваются четыре архитектуры, а экспериментально сопоставляются две реализации. Практический результат — прототипы полного обновления и варианта SCD-2, проверенные в одинаковом сценарии обработки сообщений.

Материал и методы исследования

Использованы анализ источников, сравнение архитектур, проектирование программных систем, моделирование потока изменений, реализация прототипов и экспериментальное сопоставление. Объект исследования — синхронизация продуктивного и аналитического хранилищ; предмет — способы применения вставок, обновлений и удалений и их характеристики.

Входное сообщение содержит тип операции o, ключ сущности и передаваемые поля. Значения o = i, u, d соответствуют INSERT, UPDATE и DELETE. Все поля ключа присутствуют в каждом сообщении; UPDATE изменяет переданные поля существующей записи, DELETE удаляет запись по ключу. В тестовой схеме использованы key1, key2, key3, изменяемые value1, value2, value3 и поле o. Требование доставки без потерь и дубликатов рассматривается совместно с корректностью итогового состояния.

При прямом применении запросов каждое сообщение вызывает изменение целевого хранилища. Простота этого варианта сочетается с зависимостью от поддержки и стоимости UPDATE и DELETE. Полное обновление накапливает изменения в буфере, объединяет их с текущим состоянием и перезаписывает целевую таблицу; затраты растут с объемом всей таблицы. Секционное обновление ограничивает перезапись отдельными разделами, но требует выбора ключа секционирования и организации параллельных обработчиков.

Архитектура SCD-2 предполагает хранение исторических версий с признаками актуальности [5]. Это усложняет чтение и увеличивает объем хранения. Для реализации выбраны полное обновление как более простой подход и SCD-2 как подход, позволяющий избежать полной перезаписи хранилища при каждом обновлении. В реализованном варианте SCD-2 применяющий компонент выполняет MERGE по ключу с использованием временной метки и признака удаления.

В качестве брокера выбран Apache Kafka, для обработки — Spark Structured Streaming, для распределенного хранения — HDFS [1, 2, 6]. Kafka обеспечивает хранение лога и чтение сообщений потребителями [7]. Полное обновление использует Parquet, второй вариант — Delta Lake. Прототипы написаны на Python 3.7.10 с использованием Spark 3.3.0. Docker и сервисы Kafka, Zookeeper, Schema Registry и Kafka Control Center применялись для локальной отладки; основные прогоны выполнены на распределенном кластере.

Разработаны два транслятора: отправка запросов из файла для точечной проверки и случайный генератор. Сообщения сериализуются в Avro, схема публикуется в Schema Registry. Генератор отслеживает состояния «создан» и «удален» и выбирает допустимые операции. Для экспериментального сравнения использован заранее подготовленный фиксированный набор сообщений.

Реализация применения изменений

В архитектуре полного обновления Spark Structured Streaming считывает Kafka, удаляет служебный префикс сериализации, десериализует ключи и значения по схеме и записывает поток в буфер Parquet. Отдельный модуль создает копию целевого хранилища, объединяет ее с буфером и последовательно сворачивает операции по ключу. Записи текущего состояния получают вспомогательный тип end; более ранние изменения относительно сохраненного состояния исключаются. Итоговая таблица записывается в режиме overwrite, после чего обработанные блоки буфера удаляются. Копия необходима для разделения источника чтения и места перезаписи.

Во втором варианте отдельные буфер, копия хранилища и сервис синхронизации отсутствуют. Spark Structured Streaming читает и десериализует сообщения, объединяет изменения внутри микропакета и применяет их к Delta Lake. Свертка по ключу необходима, чтобы несколько входных изменений не конкурировали за обновление одной целевой записи. Для учета наличия записи формируются updating_value и inserting_value; используется признак удаления. Помимо INSERT, UPDATE и DELETE, промежуточное представление содержит UPSERT: вставку при отсутствии записи и обновление при ее наличии.

Последовательность операций обрабатывается с учетом предыдущего состояния и заполненности полей. Удаление меняет признак наличия записи; последующая вставка может восстановить ее. Частичное обновление сохраняет ранее накопленные значения полей, не переданных в текущем запросе. После свертки MERGE сопоставляет записи по ключам, а условия whenMatchedUpdate и whenNotMatchedInsert определяют обновление, логическое удаление или вставку.

Условия и методика эксперимента

Кластер включал три узла, каждый с 128 CPU, 256 GB RAM и 5 TB SSD; суммарно — 384 CPU, 768 GB RAM и 15 TB SSD. Обе реализации запускались с одинаковой базовой конфигурацией Spark (табл. 1). Часть ресурсов оставалась для Kafka, HDFS, системных процессов и сетевого обмена. В Kafka использовалась репликация между узлами, в HDFS — репликация блоков.

Таблица 1. Параметры Spark и чтения Kafka

Параметр	Значение
Executors / ядра на executor / всего ядер	18 / 8 / 144
Память / memory overhead на executor	24 GB / 4 GB
Память driver	16 GB
Динамическое выделение ресурсов	Отключено
Shuffle partitions / default parallelism	576 / 576
maxOffsetsPerTrigger	400 000 сообщений
Режим / начальные offsets	Микропакеты / earliest
Партиционирование Kafka	По ключу сущности

Перед каждым прогоном в Kafka загружался один и тот же набор из 10 000 000 сообщений: 3 000 000 INSERT (30 %), 5 000 000 UPDATE (50 %) и 2 000 000 DELETE (20 %). Во время измерений новые сообщения не поступали. Перед стартом очищались checkpoint, временные каталоги, буферные таблицы и целевое хранилище; чтение начиналось с начальных offsets. Для каждой реализации выполнено по 10 прогонов до исчерпания лога.

Пропускная способность определялась как число входных сообщений, деленное на время полного завершения обработки накопленного лога. Задержка — среднее время от считывания микропакета из Kafka до фиксации результата в аналитическом хранилище. Эта метрика не включает ожидание в очереди при непрерывном поступлении данных и не является сквозной задержкой доставки. По десяти прогонам рассчитаны среднее, стандартное отклонение (SD), минимум, максимум и 95 % доверительный интервал среднего: x̄ ± t · s / √n, где n = 10, s — SD, t ≈ 2,262 — критическое значение распределения Стьюдента с 9 степенями свободы.

Результаты исследования и их обсуждение

Таблица 2. Производительность, по 10 прогонов

Показатель	Полное обновление	SCD-2
Задержка, с: среднее ± SD	41,8 ± 3,1	4,9 ± 0,4
Минимум–максимум, с	37,0–47,0	4,3–5,6
95 % ДИ среднего, с	39,6–44,0	4,6–5,2
Пропускная способность, записей/с: среднее ± SD	14 800 ± 940	76 800 ± 4 300
Минимум–максимум, записей/с	13 200–16 100	69 800–83 500
95 % ДИ среднего, записей/с	14 100–15 500	73 700–79 900

Вариант SCD-2 показал меньшую задержку и большую пропускную способность (табл. 2). Интервалы средних значений не пересекаются. Различие согласуется с тем, что полное обновление требует переноса всего текущего состояния, тогда как второй вариант применяет результат микропакета без полной перезаписи таблицы. Вывод относится к данным прототипам, конфигурации кластера и выбранному профилю операций; он не устанавливает универсального превосходства для любых объемов и режимов нагрузки.

Проверка восстановления после сбоев

Контрольный прогон без отказов выполнялся на том же наборе из 10 млн сообщений. Моделировались остановка Spark driver, остановка одного executor, недоступность Kafka broker в течение 60 с, недоступность узла HDFS/DataNode и остановка приложения при overwrite или MERGE. По методике время восстановления отсчитывалось от возникновения сбоя до первого успешно завершенного микропакета после восстановления компонента или перезапуска приложения.

Корректность оценивалась по совпадению итогового состояния с эталоном, отсутствию потерянных и повторно примененных операций. В процедуре контроля использовались operation_id, kafka_partition и kafka_offset; дополнительно проверялись актуальные состояния по ключу, а для SCD-2 — непротиворечивость интервалов версий. Результаты приведены в табл. 3. Во всех указанных сценариях зарегистрированы нулевые потери и дубли, итоговые состояния совпали с эталоном.

Таблица 3. Время восстановления в проверенных сценариях

Сценарий	Полное обновление, с	SCD-2, с
Сбой Spark-приложения	94	51
Сбой Spark executor	38	27
Недоступность Kafka broker	72	48
Недоступность узла хранилища	116	83
Сбой во время записи результата	128	91

Полученные результаты характеризуют выполненные проверки, но не являются общей гарантией exactly-once. Checkpoint позволяет возобновить обработку, а транзакционный журнал Delta Lake участвует в фиксации результата. При этом корректность повторного применения определяется также логикой записи.

Трудоемкость и условия выбора

Трудоемкость оценивалась как суммарное активное время проектирования, реализации, интеграции, отладки и проверки прототипа. Изучение литературы, оформление текста и пассивное ожидание прогонов не учитывались. Оценки получены суммированием этапов (табл. 4) и характеризуют конкретную разработку, а не норматив времени.

Таблица 4. Приближенная трудоемкость реализации

Вид работ	Полное обновление, чел.-ч	SCD-2, чел.-ч
Проектирование логики изменений	6	8
Настройка окружения и интеграции	5	6
Генератор тестовых данных и схемы	4	4
Чтение и десериализация Kafka	6	5
Хранение и применение изменений	14	52
Проверка хранилищ и ограничений формата	0	35
Тестирование, отладка, восстановление	5	10
Итого	40	120

Основной прирост трудозатрат второго варианта связан с логикой применения изменений и проверкой ограничений хранилищ. Обе реализации обрабатывают изменения в рамках заданной схемы сообщений; испытание произвольной эволюции схемы из этого не следует. Полное обновление целесообразно для небольших систем с ограниченной частотой изменений и допустимой задержкой обновления. Для значительных объемов и более жестких требований к задержке результаты поддерживают выбор варианта SCD-2 при готовности к большей сложности реализации и сопровождения.

Выводы

Рассмотрены четыре архитектуры синхронизации аналитического хранилища и реализованы два подхода на Kafka и Spark Structured Streaming. При обработке заранее накопленных 10 млн сообщений полное обновление обеспечило среднюю задержку 41,8 с и пропускную способность 14 800 записей/с; вариант SCD-2 — 4,9 с и 76 800 записей/с. Приближенная трудоемкость составила соответственно 40 и 120 человеко-часов. В проверенных сценариях отказов итоговые состояния совпали с эталоном, потерь и дублей не зафиксировано.

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

Список литературы

1. Maas G., Garillot F. Stream Processing with Apache Spark: Mastering Structured Streaming and Spark Streaming. O’Reilly Media, 2019. ISBN 9781491944240.

2. Armbrust M., Das T., Torres J., Yavuz B., Zhu S., Xin R., Ghodsi A., Stoica I., Zaharia M. Structured Streaming: A Declarative API for Real-Time Applications in Apache Spark // Proceedings of the 2018 International Conference on Management of Data. 2018. P. 601–613. DOI: 10.1145/3183713.3190664.

3. Shahverdi E., Awad A., Sakr S. Big Stream Processing Systems: An Experimental Evaluation // 2019 IEEE 35th International Conference on Data Engineering Workshops (ICDEW). 2019. P. 53–60. DOI: 10.1109/ICDEW.2019.00-35.

4. Puentes F., Pérez-Godoy M. D., González P., del Jesus M. J. An analysis of technological frameworks for data streams // Progress in Artificial Intelligence. 2020. Vol. 9. No. 3. P. 239–261. DOI: 10.1007/s13748-020-00210-6.

5. Beryoza D., Campbell M., Cardorelle C., Creasey T., Cushing D., Da Silva V., David S., Hagleitner A., Henderson I., Howell D., Kozine I., Prieto P., Thompson P., Vazquez J., Zhang Y. IBM Cognos Dynamic Cubes. 2nd ed. IBM Redbooks, 2015. SG24-8064-01. URL: https://www.redbooks.ibm.com/abstracts/sg248064.html (дата обращения: 29.10.2026).

6. Shvachko K., Kuang H., Radia S., Chansler R. The Hadoop Distributed File System // 2010 IEEE 26th Symposium on Mass Storage Systems and Technologies (MSST). 2010. P. 1–10. DOI: 10.1109/MSST.2010.5496972.

7. Bejeck W. P. Jr. Kafka Streams in Action: Real-time apps and microservices with the Kafka Streams API. Manning Publications, 2018. 280 p. ISBN 9781617294471.

Код для цитирования: Скопировать