Обработка запросов в системе с лямбда-архитектурой на уровне ускорения
Ю.А. Григорьев, д-р техн. наук,
О.Ю. Ермаков
(Московский государственный технический университет им. Н.Э. Баумана)
Аннотация
Выполнен анализ процессов потоковой обработки данных на уровне ускорения, включающий звено сбора данных, звено очереди сообщений, звено анализа, хранилище данных в памяти и звено доступа к данным. Рассмотрен алгоритм Count-Min Sketch для подсчета частоты и суммы значений какого-либо элемента в потоке. Показано, что использование эскиза (sketch) приводит к большой ошибке восстановления накопленных значений при достаточно большом числе элементов в потоке. Предложена реализация звена анализа на уровне ускорения в системе с лямбда- архитектурой с плавающим окном. Звено включает матрицу векторов (одномерных числовых массивов) вместо эскизов. Это позволяет читать накопленные значения из векторов матрицы напрямую.
Ключевые слова: лямбда-архитектура, потоковая обработка, уровень ускорения, эскиз, вектор. потоковый данные алгоритм
Введение
Обработка больших объемов данных в реальном времени является важным требованием в современных высоконагруженных системах. Для решения этой задачи используются системы потоковой обработки данных. Обработка потоков данных нашла применение в различных областях: в поисковых системах, в социальных сетях, а также в системах обнаружения мошенничества в торговых и финансовых системах, в системах контроля состояния оборудования, военных и разведывательных системах [1].
Одним из вариантов реализации потоковой обработки данных является лямбда-архитектура [2]. В [3, 4] представлена реализация лямбда-архитектуры для создания бэкэнда обработки данных в Amazon EC2, обеспечивающая высокую пропускную способность при невысокой стоимости обслуживания сети. Лямбда-архитектура используется для реализации потоковой обработки во многих других областях: отслеживание тепловой карты (неаішар) [5], обработка запросов [6], в медицине для внутри хирургических прогнозов [7] и др.
Лямбда-архитектура имеет пакетный уровень и уровень ускорения. На уровне пакетной обработки хранится главная копия массива данных. На основе этих данных формируются пакетные представления (уровень обслуживания), которые обеспечивают быстрый доступ к сущностям и к интегрированным показателям, полученным за определенный промежуток времени. Уровень ускорения обеспечивает обработку данных в реальном масштабе времени, так как требуемые данные не могут быстро появиться на уровне обслуживания. В работе [8] показано, что существующая схема лямбда-архитектуры имеет ряд существенных недостатков. В частности, если требуется новое пакетное представление, то для его создания необходимо выполнить поиск по всей большой базе данных пакетного уровня, т.е. для каждого нового запроса требуется выполнить поиск в большой базе данных.
В работе [8] была предложена новая блок-схема лямбда-архитектуры (рис. 1).
Рис. 1. Новая блок-схема лямбда-архитектуры [8].
Метаданные используются для приближенного вычисления агрегированных значений (sum, avg, count) необходимых атрибутов JSON- документов. Для реализации поискового запроса требуется n " N случайных чтений сегментов базы данных, где N - общее число сегментов, т.е. скорость выполнения запросов существенно увеличивается при незначительной потере точности вычислений агрегированных значений. Уменьшение ошибки оценки этих значений достигается за счет нового метода расчета вероятностей чтения сегментов.
В статье предлагаются решения потоковой обработки данных в системе с лямбда-архитектурой на уровне ускорения.
Анализ процессов потоковой обработки данных на уровне ускорения
В [9] предлагается целостный подход к организации потоковой обработки данных, который ориентирован, в основном, на реализацию уровня ускорения. Соответствующая архитектурная диаграмма включает следующие звенья:
звено сбора данных;
звено очереди сообщений;
звено анализа;
хранилище данных в памяти;
звено доступа к данным.
Кратко рассмотрим перечисленные звенья потоковой обработки данных на уровне ускорения.
1. Звено сбора данных. Здесь используется паттерн "поток". Данные поступают от мобильных устройств (или средств), они предварительно сохраняются в лог-журналах с целью увеличения надежности системы (протоколирование с использованием методов RBML, SBML, HML) и затем передаются на вход следующего звена. Например, данные о выполненных заказах такси непрерывно поступают в систему и там обрабатываются.
2. Звено очереди сообщений. В качестве примера можно указать следующие средства обмена сообщениями: NSQ, ZeroMQ, Apache Kafka. Одним из самых популярных решений является проект Apache Kafka [10], отличающийся от аналогов своей надежностью и предоставлением семантики exactly-once [11]. Он позволяет публиковать потоки сообщений и подписываться на них.
В этом звене можно выделить три главных компонента: производитель (звено сбора данных), брокер, потребитель (звено анализа). На рис. 2 приведена схема обмена сообщениями, где буквой "Б" обозначен компонент брокера. От звена сбора данных поступают сообщения, которые брокер ставит в очередь и затем по подписке передает ("проталкивает") их брокеру- получателю. Тот ставит сообщения в выходную очередь. Звено анализа посылает запросы брокеру и читает ("вытягивает") сообщения из очереди.
Рис. 2. Схема работы брокера.
В зависимости от реализации брокера "проталкивание" может быть заменено на "вытягивание" и наоборот. Брокеры объединены в логический кластер. Параметры звена очереди сообщений должны быть подобраны так, чтобы не было переполнения очередей. Для этого работа брокеров должна быть промоделирована с помощью системы массового обслуживания [12, 13].
После некоторой фильтрации полученные сообщения сохраняются в журнале изменений (см. рис. 1) для их дальнейшей обработки на пакетном уровне и уровне обслуживания (метаданные). Также эти данные поступают в звено анализа уровня ускорения.
3. Звено анализа. В настоящее время существует немало технологий анализа данных. Самыми популярными из продуктов с открытым исходным кодом являются Spark Streaming, Storm, Flink и Samza [14 - 16]. Все они - проекты Apache. Перечисленные системы обладают рядом общих черт [9] (рис. 3).
Потоковый диспетчер распределяет приложения анализа (см. ниже) по потоковым процессорам распределенной системы. Сообщения, поступающие из звена очереди сообщений, объединяются в пакеты, которые накапливаются в системе в течение некоторого интервала времени А. Далее потоковый диспетчер распределяет пакеты по потоковым процессорам, их обрабатывают приложения анализа. Важно, чтобы обработка завершилась за время, меньшее А. Потоковый процессор называется по-разному: в Spark Streaming - это "исполнитель Spark", в Storm - "супервизор", в Flink - "исполнитель", в Samza - "исполнитель заданий Samza".
Рис. 3. Общая схема звена анализа.
Приложения анализа могут быть различными:
подсчет уникальных значений (на основе битовых комбинаций, например, алгоритмы LogLog, HyperLogLog, HyperLogLog++ [17, 18], или на основе порядковых статистик, например, алгоритм MinCount [19]);
подсчет частоты и суммы значений какого-либо элемента в потоке (например, алгоритм Count-Min Sketch [20]);
определение, встречалось ли значение в потоке ранее (алгоритм на основе фильтра Блума [21, 22]);
и другое
4. Хранилище данных в памяти. Для поступающих элементов вычисляются хеш-функции, и полученные значения накапливаются (или обновляются) в таблице каждого потокового процесса (рис. 4). Перечисленные в пункте 3 приложения анализа обладают свойством линейности: результирующую таблицу можно получить путем простого сложения (или обновления) локальных таблиц. Вычисление хеш-функций, накопление или обновление таблиц выполняются относительно быстро. Объем каждой таблицы небольшой и составляет несколько килобайтов, поэтому передача их по сети занимает немного времени. Но анализ показывает (см. следующий раздел), что погрешность воспроизведения (по запросу) значений исходных элементов по результирующей таблице может быть достаточно большой.
5. Звено доступа к данным. Существует много паттернов взаимодействия потокового клиента (получателя данных) с хранилищем данных [9]: синхронизация данных (Data Sync), удаленный вызов метода или процедуры (RMI/RPC), простой обмен сообщениями, издатель-подписчик. При этом также необходимо выбрать протокол отправки данных клиентам [9]: вебуведомления (webhook), длинный HTTP-опрос, протокол пересылаемых сервером событий (Server-Sent Events, SSE), веб-сокеты (WebSocket). Протокол WebSocket (существует с 2011 г.) превосходит по характеристикам остальные протоколы.
Рис. 4. Схема формирования хранилища данных в памяти.
Алгоритм Count-Min Sketch для подсчета частоты и суммы значений какого-либо элемента в потоке
Подсчет частоты и суммы значений элементов, появляющихся в потоке, является фундаментальной задачей во многих приложениях потоков данных - таких как отслеживание финансовых данных, обнаружение вторжений, мониторинг сети [23], обработка сообщений от мобильных устройств и средств, торговых центров и др.
Для решения этой задачи был разработан алгоритм Count-Min Sketch [20]. Он стал одним из первых для целого класса подобных алгоритмов. В [25] приведена теория распределения эскизов по узлам, учитывающая их свойство линейности. Общая теория эскизов изложена в [26], где также имеются рекомендации по выбору хеш-функций (с. 219).
Рассмотрим алгоритм Count-Min Sketch более подробно [20].
1. Структура данных.
Эскиз (sketch) представлен двумерным массивом (таблицей) count[d,w], где d - число строк, w - число столбцов. Заданы параметры (є, ф) и w = Te / є! и d = Tin (1 / ф)l, e - основание натурального логарифма. Все элементы массива (ячейки таблицы) первоначально равны нулю. Кроме того, заданы d хэш-функций:
hi . . . hd : {1 . . . n} ^ {1 . . . w}. (1)
Будем считать, что hk(i) - случайная целочисленная величина, которая равномерно распределена на отрезке [1,w] для каждого i=\...n. Также предполагается, что [hk(i)}k независимы для каждого i. Независимость сохраняется и по i.
2. Обновление эскиза.
Пусть из потока поступает пара (i, ci), где i - номер элемента, Ci > 0 - его значение (если c=1, то эскиз используется для подсчета числа появлений этого элемента в потоке, т.е. частоты). К некоторой ячейке каждой строки таблицы добавляется величина c (рис. 5). Формально это можно записать так:
count[k, hk(i)]<-- count[k, hk(i)]+ Ci, k = 1,..., d. (2)
Рис. 5. Схема обновления эскиза.
3. Чтение (восстановление)накопленных значений Ci элемента i (ai*).
Восстановленное значение рассчитывается по формуле:
at* = min k count[k, hk(i)]. (3)
4. Оценка точности восстановленного значения ai*.
Границы полученной оценки а* имеют следующие значения [20]:
ai< a*; ai* < at + є||а||1 - с вероятностью не менее 1 - Ф, (4)
где ai - точное значение накопления; ||а||1 - метрика L1.
Для получения правой границы ai* было использовано неравенство Маркова. Она может быть большой, все зависит от накопленных величин L1 = 'Lai (см. (4)). В [24] предлагается уменьшить значение метрики L1 путем вычитания из a некоторого вектора Я с одинаковыми значениями элементов. Для определения Ятребуется оценить медиану точных величин {a,} по некоторой случайной выборке, которую надо как -то получить. Здесь также для некоторых распределений { ai} правая граница ai* может быть большой.
Оценим точность восстановленного значения иначе. Ясно, что если n < w-d, то использовать эскиз не имеет смысла. В этом случае выгоднее применить вектор длиной n: так как и память меньше, и сохраняются точные значения накоплений {a/}. Поэтому далее будем полагать, что n > w-d.
Оценим сначала вероятность, что восстановленное значение ai* не будет совпадать с точным значением a*. Это вероятность,
что Vk 3(i1^i) (hk(i1) = = hk(i)):
p = (1 - (1 - 1/w)n"7)d. (5)
Выражение во внешних скобках соответствует квантору 3, а степень d - квантору V.
Выполним некоторые преобразования (5):
Здесь в преобразованиях 1 и 3 был использован 2-й замечательный предел, так как обычно w > 128 (для 1) и e(n-1)/w"1 (для 3) в силу n > w-d, а d> 8.
Пусть n = w-d+1 и d=8. Тогда из (6) получим p = 0,997. Т.е. при n > w-d и при типичных значениях w > 128 и d > 8 восстановленное значение a* не будет совпадать с точным значением ai с вероятностью, почти равной 1. Оценим ошибку восстановления.
В силу свойств хеш-функций (1) накопленные значения d-'Lat равномерно заполняют ячейки матрицы эскиза (см. рис. 5). На одну ячейку в среднем приходится (d-'Lai)/(w-d) = 'Zai/w накопленных значений. Следовательно, какое-либо восстановленное значение можно оценить как:
где aA - среднее значение величин {ai}.
Относительная погрешность восстановления ai равна
(ai* - ai)/ai = (n/w)-(aA/ai) - 1. (8)
Но n/w >d, d - число хеш-функций (обычно больше 8). И если ai не превышает среднего значения, то, как следует из (8), относительная погрешность восстановления может очень большой (сотни процентов).