---

1)  Здравствуйте, меня зовут Агей Михаил, и тема моей дипломной работы: "Реализация механизма строго однократной потоковой обработки данных при работе с нереляционным хранилищем". Научный руководитель: Николаев Владимир Вячеславович

2) В современном мире потоковая обработка данных необходима для многих процессов. Например, банковские операции, заказ такси или изменение данных о товаре. При этом наличие дубли может быть критичным. Существуют разные семантики обработки данных.Не более одного раза, не менее одного раза и ровно один раз. Потоковая обработка данных реализована, например, в стандарте вычислений Apache Spaek, который работает с хранилищем YTsaurus благодаря модулю SPYT.  Проблема в том, что спарк стриминг в YTsarus имеет гарантию "не менее одного раза", то есть возможны дубли

3) Цель моей работы - обеспечение гарантии строго однократной потоковой обработки данных в Apache Spark при записи в динамические таблицы YTsaurus. А задачи представлены на слайде

4) На слайде представлена схема потоковой обработки данных.В Apache Spark есть два компонента, драйвер и исполнители.А в хранилище есть очереди и таблицы.Исполнители читают строки, делают обработку данных и записывают строки в выходную таблицу. А драйвер делает фиксацию нового смещения. Проблема в том, что во время записи данных или между записью данных и с двигом смещения может произойти какой-либо сбой.И тогда при рестарте обработки данных та же самая порция данных будет обработана и записана в выходную таблицу повторно. 

5) Изначально существовало неполное решение этой проблемы идемпотентный приемник. Это сортированная динамическая таблица, которая имеет набор ключей. Благодаря этому невозможно записать в таблицу одну и ту же строку два раза. Если это произойдет, то новая запись перезатрет старую. Проблема в том, что это решение подходит только для трансформации с отображением один к одному.  То есть нельзя сделать join, group by.  Также запись сортированной таблицы работает все медленнее при росте таблицы, потому что при каждой вставке происходит сортировка.

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

7) Но существует проблема текущей конфигурации компонента Прокси. Этот компонент необходим для выполнения любых запросов в хранилище.По умолчанию на драйвере и на каждом экзекьюторе поднимаются свои экземпляры прокси. А таблетные транзакции привязываются к конкретным Прокси.  Из-за этого, например, нельзя создать транзакцию на драйвере, а потом использовать ее исполнителями

8) Было рассмотрено два способа решения этой проблемы.  Во-первых, можно было все действия выполнять на одном исполнителе. То есть на исполнителе создается транзакция, он читает данные, он обрабатывает данные, записывает данные в выходную таблицу, и он же обновляет смещение. 

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

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

11) Поэтому было выбрано и реализовано второе решение.  Высокоуровнево механизм достаточно простой.  Драйвер создает транзакцию на случайной прокси.  Используя метаинформацию об этой транзакции, исполнители обращаются к той же самой прокси, для того, чтобы записать данные.  После этого драйвер фиксирует смещение.  Если все этапы выполнились успешно, драйвер фиксирует транзакцию. Если что-то пошло не так, драйвер отменяет транзакцию. Благодаря тому, что смещение всегда правильно указывает на последнюю обработанную строку, во время потоковой обработки не возникнет дублей и не потеряются данные

12)  Для того, чтобы проверить данную функциональность, написаны тесты. Модульные тесты, интеграционные тесты и бенчмарки. Модульные интеграционные тесты проходят, а бенчмарки показывают полезную информацию для сравнения.

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

14) Также была произведена оценка производительности. Сравнивалось время, за которое будет обработан один миллион строк, одной тестовой очереди, объем которой примерно один гигабайт. Сравнение показало, что скорость работы увеличилась примерно на 15%

15) В результате работы Реализован механизм строго-однократной потоковой обработки данных при работе с не реляционным хранилищем YTsaurus и Проведено комплексное тестирование механизма. В качестве дальнейшей работы, может быть: Оптимизация производительности транзакционного режима и 
Выявление зависимости критической нагрузки на rpc-proxy от потока данных

Цель работы достигнута!

16) Спасибо за внимание