Итак, можно сделать следующие выводы:
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.
Заключение
Показано, что использование эскиза приводит к большой ошибке восстановления накопленных значений при достаточно большом числе элементов в потоке.
Предложена структура уровня ускорения с использованием векторов для накопления значений элементов, позволяющая в динамике подключать (или исключать) векторы и хеш-таблицы для новых показателей Y и ключей X, поступающих в потоке данных. Имеется возможность в динамике строить комбинации ключей, что позволяет выполнять запросы select к векторам c использованием конструкции group by по этим комбинациям.
Показано, что при переполнении какого-либо вектора возможно уменьшение окна (плавающее окно). Но в этом случае следует учитывать, что увеличивается нагрузка X на клиента, обрабатывающего запросы к хранилищу данных в памяти.
Объем векторов, которые хранятся в ОП узла, невелик. Это позволяет быстро передавать векторы по сети и объединять их на координирующем сервере, используя свойство линейности.
Литература
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.