Обработка запросов в системе с лямбда-архитектурой на уровне ускорения

В [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)

Итак, можно сделать следующие выводы:

1) если n < w-d, то использовать эскиз не имеет смысла, в этом случае выгоднее применить вектор (одномерный числовой массив) длиной n;

2) если n > w-d, то погрешность восстановления накопленных значений {ai} может быть очень большой.

Появились работы, в которых предлагаются методы более сложной обработки потоковых данных, не связанные с эскизами. В публикациях [27, 28] рассматриваются методы, позволяющие обнаружить выбросы в данных (например, большое время подачи машины, много отказов для конкретного водителя, отсутствие машины определенного класса и др.). В работе [29] предлагается модель конвоя (convoy), позволяющая на основе анализа потока данных выявить, например, скопление более q машин на расстоянии меньше esp от каждой за время t. То же можно сделать для скопления людей, которым требуется подать автомобили, и др.

Предложения по реализации звена анализа в системе с лямбда-архитектурой

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

Из звена очереди сообщений поступают записи <ключи, показатели>. По ключу или их комбинации из соответствующей хеш-таблицы читается номер i, который используется для обновления всех векторов (по смещению 1-ї), соответствующих показателям. Для чтения накопившихся в векторах в течение окна значений (частот, времени и др.) используется select-подобный оператор. Тренды этих значений отображаются на экране, и оператор может выявить в какие-то моменты времени критические ситуации.

Для иллюстрации изложенного подхода рассмотрим предметную область "Обслуженные заказы такси". Из потока поступают записи <ключи, показатели>.

I. В качестве показателей (Y) выступают следующие атрибуты (в скобках указаны возможные значения):

1 - заказ обслужен (1 или 0);

2 - время подачи (время подачи машины с момента поступления заказа);

3 - отказ пассажира (1 или 0);

4 - отказ водителя (1 или 0);

5 - машина попала в дорожно-транспортное происшествие (ДТП) (1 или 0);

6 - машина остановлена дорожно-постовой службой (ДПС) (1 или 0);

7 - отрицательный отзыв пассажира (1 или 0).

II. В качестве ключей и их комбинаций (X) используются следующие атрибуты:

1 - водитель - ключ;

2 - район подачи (машины) - ключ;

3 - (водитель, район подачи) - комбинация ключей,

4 - класс машины - ключ;

5 - номер машины - ключ;

6 - (водитель, номер машины) - комбинация ключей.

III. Имеется несколько хеш-таблиц для ключей и их комбинаций - <ключ, значение>:

1 - <водитель, номер i для всех vector 1.Y>, Y=1..7;

2 - <район подачи, номер i для всех vector 2.Y>, Y=1..7;

3 - <(водитель, район подачи), номер i для всех vector 3.Y>, Y=1..7;

4 - <класс машины, номер i для всех vector 4.Y>, Y=1..7;

5 - <номер машины, номер i для всех vector 5.Y >, Y=1..7;

6 - <(водитель, номер машины), номер i для всех vector 6.Y>, Y=1..7.

IV. Обновление векторов vector X.Y (рис. 6).

Рис. 6. Схема обновления vector X.Y.

В потоке поступает выполненный заказ <ключи, показатели>. Из него выделяются значения ключей: водитель, район подачи, класс машины, номер машины (X = 1, 2, 4, 5). Строятся две комбинации ключей: (водитель, район подачи) и (водитель, номер машины) (X = 3, 6). Из хеш-таблицы X (X = 1...6) по значению ключа хеш-таблицы читается номер ix для vector X.Y. Номер ix используется для обновления ячеек (по смещению ix - 1) всех vector X.Y для данного X (Y = 1..7). При этом в зависимости от значения показателя Y к ячейке добавляется (суммируется) либо величина показателя, либо ничего не добавляется (если 0). Если в хеш-таблице X нет соответствующей записи, то она включается и присваивается номер i (следующий по порядку), который затем используется для поиска в векторах vector X.Y, Y = 1...7 (номер i должен быть уникальным для хеш-таблицы X).

V. Примеры запросов.

1. Найти среднее время подачи такси водителем в каком-нибудь районе:

select водитель, район подачи, время подачи/заказ обслужен as mean

from vector 3.1, vector 3.2

group by водитель, район подачи.

Просматриваются все записи хеш-таблицы 3 (см. рис. 6), для каждого ключа "водитель, район подачи" читается номер i = i3. Этот номер используется для чтения значений "время подачи" (из vector 3.2) и "заказ обслужен" (из vector 3.1). Выполняется деление этих величин.

2. Вывести все показатели для класса машины "Эконом":

select *

from vector 4.*

where класс машины= "Эконом".

Из хеш-таблицы 4 (класс машины) читается соответствующий номер i = i4. Этот номер используется для чтения значений из vector 4.Y, Y = 1...7.

Запросы выполняются в моменты перемещения окна. Т.е. читаются значения показателей, собранные за интервал времени окна. При этом используется следующий алгоритм:

1) положить T=0, W=Wi=Wo - начальный размер окна (с течением времени может быть плавающим);

2) обнулить все хеш-таблицы и векторы vector X.Y;

3) в момент времени t = T + W окончания окна (размер текущего окна положить равным W = W1 = W0) или когда номер элемента в потоке больше n (размер текущего плавающего окна положить равным W = t - T) активизируется программа, с помощью которой выполняются запросы, отображаются результаты текущего окна, эти значения добавляются к предыдущим результатам для получения трендов:

если W1 - W > 0, то положить W=W1=W1 - W, // новое плавающее окно T = t;

4) перейти к пункту 2 алгоритма.

Для доступа к хранилищу данных в памяти можно использовать Web- сокеты (см. звено доступа к данным). После получения от клиента команды "притормози" (клиент перегружен) можно автоматически увеличить размер окна (уменьшить нагрузку X на клиента). Но при этом следует учитывать, что количество номеров элементов в потоке может стать больше размера вектора n (см. предыдущий алгоритм).

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

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

Объем векторов, которые хранятся в ОП узла, невелик. Предположим, что n = w-d = 27-23 = 210 (размер одного эскиза). Объем одного вектора равен v1 = n-4 (байтов) = 4КБ. Пусть число хеш-таблиц равно 6 (число ключей и их комбинаций), а число показателей равно 7. Тогда объем всех векторов в ОП узла равен V = 4(КБ)-6-7 = 168 КБ.

Можно выделить следующие преимущества предложенного подхода к реализации звена анализа уровня ускорения:

в динамике можно подключать (или исключать) векторы для новых показателей Y ;

в динамике можно подключать (или исключать) хеш-таблицы с новыми ключами или их комбинациями (X);

имеется возможность строить комбинации ключей, что позволяет выполнять запросы select c group by по этим комбинациям;

можно использовать плавающее окно, если число разных элементов в потоке станет больше n. Это позволяет экономить память и время обновления вектора, поскольку нет необходимости увеличивать его размер. Размер вектора следует динамически увеличить только в случае перегрузки клиента, выполняющего доступ к хранилищу данных в памяти - и то, если при увеличении размера окна число элементов в потоке становится больше n.

Заключение

Литература

1. Клепплан М. Высоконагруженные приложения. Программирование, масштабирование, поддержка. - СПб.: Питер, 2018.

2. Марц Н., Уоррен Д. Большие данные: принципы и практика построения масштабируемых систем обработки данных в реальном времени. - М.: ООО "И.Д. Вильямс", 2016.

3. Kiran M. et al. Lambda architecture for cost-effective batch and speed big data processing // 2015 IEEE International Conference on Big Data (Big Data). IEEE. - 2015. - Р. 27852792.

4. Gribaudo M., Iacono M., Kiran M. A performance modeling framework for lambda architecture based applications // Future Generation Computer Systems. - 2018. - Vol. 86. - Р. 1032-1041.

5. Perrot A. et al. HeatPipe: High Throughput, Low Latency Big Data Heatmap with Spark Streaming // 2017 21st International Conference Information Visualisation (IV). IEEE. - 2017. - Р. 66-71.

6. Yang F. et al. The RADStack: Open source lambda architecture for interactive analytics // Proceedings of the 50th Hawaii International Conference on System Sciences. - 2017. - P. 1703-1712.

7. SpangenbergN., Wilke M., FranczykB. A Big Data architecture for intra-surgical remaining time predictions // Procedia computer science. - 2017. - Vol. 113. - Р. 310-317.

8. Григорьев Ю.А., Ермаков О.Ю. Лямбда-архитектура системы с уровнем обслуживания на основе метаданных для приближенной обработки запросов // Информатика и системы управления. - 2019. - №2. - С. 3-15.

9. Пселтис Э.Д. Потоковая обработка данных. Конвейер реального времени. - М.: ДМК Пресс, 2018.

10. Apache Kafka. A distributed streaming platform: https://kafka.apache.org/ (дата обращения: 31.03.2020).

11. Нархид Н., Шапира Г., Палино Т. Apache Kafka. Потоковая обработка и анализ данных. - СПб.: Питер, 2018.

12. Wu H., Shang Z., Wolter K. Performance Prediction for the Apache Kafka Messaging System // 2019 IEEE 21st International Conference on High Performance Computing and Communications; IEEE 17th International Conference on Smart City; IEEE 5th International Conference on Data Science and Systems (HPCC/SmartCity/DSS). IEEE. - 2019. - Р. 154-161.

13. KroЯ J., Krcmar H. Modeling and simulating apache spark streaming applications //Softwaretechnik-Trends. - 2016. - Vol. 36, №. 4. - Р. 1-3.

14. Quoc D.L. et al. Approximate stream analytics in apache flink and apache spark streaming // arXiv preprint arXiv:1709.02946. - 2017.

15. Chintapalli S. et al. Benchmarking streaming computation engines: Storm, flink and spark streaming // 2016 IEEE international parallel and distributed processing symposium workshops (IPDPSW). IEEE. - 2016. - Р. 1789-1792.

16. Noghabi S.A. et al. Samza: stateful scalable stream processing at LinkedIn //Proceedings of the VLDB Endowment. - 2017. - Vol. 10, №. 12. - Р. 1634-1645.

17. Flajolet P. et al. Hyperloglog: the analysis of a near-optimal cardinality estimation algorithm // Conference on Analysis of Algorithms. - 2007. - P.127-146.

18. Heule S., Nunkesser M., Hall A. HyperLogLog in practice: algorithmic engineering of a state of the art cardinality estimation algorithm // Proceedings of the 16th International Conference on Extending Database Technology. - 2013. - P. 683-692.

19. Giroire F. Order statistics and estimating cardinalities of massive data sets // International Conference on Analysis of Algorithms DMTCS proc. AD. - 2005. - Vol. 157. - P. 166.

20. Cormode G., Muthukrishnan S. An improved data stream summary: the count-min sketch and its applications // Journal of Algorithms. - 2005. - Vol. 55, №. 1. - P. 58-75.

21. Bloom B.H. Space/time trade-offs in hash coding with allowable errors // Communications of the ACM. - 1970. - Vol. 13, № 7. - P. 422-426.

22. Tarkoma S., Rothenberg C.E., Lagerspetz E. Theory and practice of bloom filters for distributed systems. // IEEE Communications Surveys and Tutorials. -2012. - № 14(1). - P.131-155.

23. Basat R. B., Friedman R., Shahout R Stream frequency over interval queries // Proceedings of the VLDB Endowment. - 2018. - Vol. 12, №. 4. - P. 433-445.

24. Chen J., Zhang Q. Bias-Aware Sketches // Proceedings of the VLDB Endowment. - 2017. - Vol. 10, № 9. - P. 961-972.

25. Cormode G., Garofalakis M. Sketching streams through the net: Distributed approximate query tracking // Proceedings of the 31st international conference on Very large data bases. - 2005. - P. 13-24.

26. Cormode G. et al. Synopses for massive data: Samples, histograms, wavelets, sketches // Foundations and Trends® in Databases. - 2011. - Vol. 4. - № 1-3. - С. 1-294.

27. Yoon S., Lee J.G., Lee B.S. NETS: extremely fast outlier detection from a data stream via set-based processing // Proceedings of the VLDB Endowment. - 2019. - Vol. 12, №. 11. - P. 1303-1315.

28. Cao L. et al. Efficient discovery of sequence outlier patterns // Proceedings of the VLDB Endowment. - 2019. - Vol. 12, №. 8. - С. 920-932.

29. Orakzai F., Calders T., Pedersen T.B. k/2-hop: fast mining of convoy patterns with effective pruning // Proceedings of the VLDB Endowment. - 2019. - Vol. 12, №. 9. - P. 948960.

Размещено на Allbest.ru