Программные продукты и системы

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

За последнее время было разработано большое количество систем [2–5], обрабатывающих большие графы на одном компьютере  с приемлемой производительностью. Однако  в основном они плохо подходят для ситуаций, когда структура графа неизвестна заранее, а также существенно изменяется в процессе работы.

Большинство таких систем, включая GraphChi [2], TurboGraph [3], VENUS [4], требуют специального преобразования графа в представление, подходящее для эффективной обработки. Это преобразование необходимо повторять при существенных изменениях структуры графа, что ведет к значительным накладным расходам. Графовые БД [6], являющиеся наиболее подходящими системами для работы с динамическими графами, в целом имеют невысокую скорость доступа к данным. Доработанные системы обработки графов на отдельном компьютере, такие как GraphChi-DB [7], также показывают невысокую производительность.

Далее описаны ключевые детали реализаций наиболее популярных систем, экспериментальные оценки производительности и потенциальные направления для исследований.

Системы обработки больших графов  на одном компьютере

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

В основу системы положен алгоритм параллельных скользящих окон (parallel sliding windows). На этапе подготовки данных список вершин V графа G = (V, E) разбивается по порядку на P непересекающихся интервалов.  С каждым интервалом ассоциирован сегмент (shard), в котором хранятся все ребра, входящие в вершины этого интервала и упорядоченные по исходящей вершине. Интервалы выбираются так, чтобы каждый сегмент полностью помещался в оперативную память. При выполнении алгоритма происходит последовательная обработка интервалов. Для этого загружается сегмент, содержащий входящие ребра для вершин интервала, а также окна в других сегментах, содержащие исходящие ребра для интервала (рис. 1). Таким образом, при обработке каждого интервала выполняется не более P последовательных чтений. Аналогично после обработки каждого интервала происходят полная перезапись связанного с ним сегмента и обновление содержимого окон в других сегментах.

Для алгоритма существует теоретическая оценка количества операций ввода:

где |E| – количество ребер; B – размер блока, который читается/пишется на диск за один раз; P – количество сегментов, на которые разбиты данные. При этом количество непоследовательных операций ввода-вывода – Q(P2). Оценка количества операций ввода-вывода для GraphChi, а также для алгоритмов, описанных далее, выполняется на основе подхода, предложенного в [8].

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

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

Система TurboGraph [3] является попыткой решить следующие проблемы GraphChi: перед началом обновления сегмента необходимо полностью загрузить сегмент в память; все ребра из одного интервала обрабатываются последовательно, что снижает параллельность вычислений; даже при необходимости обработки небольшой части всего графа GraphChi читает весь граф на первой итерации.

Таблица 1

Время предобработки графа в сравнении  со временем выполнения для некоторых  графов и алгоритмов

Table 1

Graph preprocessing time compared  to computation time for some graphs  and algorithms

Данные хранятся в страницах со слотами, имеющих размер, кратный 1 Мб. Каждая страница содержит последовательно хранящиеся записи и их слоты в конце файла, каждый  слот – пару из ID вершины и смещения начала записи от начала файла, записи внутри страницы – списки смежности для соответствующих вершин. Идентификатор записи (RID, Record ID) для вершины состоит из идентификатора страницы и номера слота. Так как списки смежности содержат RID, в памяти хра- нится RID-таблица, позволяющая вычислить номер вершины по RID. Эта таблица хранит по одной записи для каждой страницы, содержащей номер первой вершины на ней. Вычисление номера вершины по RID происходит следующим образом: RIDTable[pageID].start- VertexID + slotNo. В большинстве случаев для списка смежности для вершины достаточно одной страницы, однако могут присутствовать вершины с очень большими списками смежности, которые занимают несколько страниц. Для корректной обработки таких вершин добавлена еще одна таблица со списком страниц больших записей (LRPL, Large Record Page List), содержащая списки страниц для каждой такой вершины. Ссылки на записи в LRPL и количество записей являются дополнительными полями в RID-таблице. На риcунке 2 изображен пример структур данных TurboGraph на диске, где vi – вершины, pj – страницы, rk – ребра.

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

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

Система X-Stream [9] использует подход, основанный на потоковой обработке неупорядоченных списков ребер. Авторы показывают, что такой подход позволяет получать производительные алгоритмы.

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

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

Итерация вычисления состоит из трех этапов.

Этап 1. Для каждого сегмента осуществляется проход по всем вершинам и формируются обновления, связанные с исходящими из вершин сегмента ребрами. Обновления записываются в общий файл обновлений для этой итерации.

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

Этап 3. Обновления применяются к значениям вершин в каждом сегменте.

Производительность X-Stream сравнивается с различными системами, в том числе с GraphChi, и существенно ее превосходит. Авторы не рассматривают динамические графы, но следует отметить, что отсутствие необходимости упорядочивать ребра позволяет доста- точно легко добавлять их в существующий граф.

Развитием GraphChi для работы с динамическими графами является система Graph- Chi-DB [7], авторы которой предлагают новую структуру данных Partitioned Adjacency Lists (PAL). Аналогично GraphChi при построении этой структуры список вершин разбивается на непересекающиеся интервалы, каждому интервалу соответствует раздел с ребрами графа, входящими в вершины этого интервала, а ребра упорядочены по номеру вершины источника. Размеры интервалов могут различаться и выбираются так, чтобы каждый раздел с ребрами мог поместиться в оперативную память. Структура файла для раздела с ребрами, используемого в системе, основана на сжатом хранении строкой [10], однако система позволяет использовать любые другие удобные форматы. Разделы с ребрами являются неизменяемыми, поэтому каждому разделу соответствует буфер, в котором накапливаются новые ребра для этого раздела аналогично GraphChi. Структура для управления разделами основана на LSM-дереве, на верхнем уровне которого хранятся буфера ребер в памяти, а на последующих – разделы на диске, при этом информация о состоянии конкретного раздела находится на нескольких уровнях дерева (рис. 3).

Для вычислений используется модель Parallel Sliding Windows, введенная в GraphChi.

Производительность системы сравнивается c MySQL и Neo4J на системе с SSD-диском. Сравнение производится на задачах выборки и вставки отдельных ребер и вершин, а также выборки окрестности вершины радиуса 2, состоящей из «друзей друзей». Несмотря на то, что в большинстве сценариев работы GraphChi-DB превосходит конкурентов, время выборки данных для нее существенно превышает аналогичное время для систем со статическим или почти статическим разбиением графа на разделы.

Альтернативные системы. Существует некоторое количество распространенных систем, массово используемых для обработки больших графов, которые не предполагают работу на одном компьютере. Однако во многих случаях такие системы возможно запустить на одном компьютере. При этом их существенным преимуществом является эффективная масштабируемость на более чем один компьютер. Типичной системой такого класса является Google Pregel [11] или его открытая реализация Apache Giraph [12]. В этой системе используются распределение вершин по узлам кластера и передача сообщений с обновлениями состояния между узлами. Работа с динамическими графами в Pregel [11] приводит либо к увеличению потока сообщений между узлами, либо к необходимости перераспределения вершин на некоторых итерациях. Также представляет интерес система Apache GraphX [13], которая использует распределенные таблицы Apache Spark [14] для хранения вершин и ребер. Такой подход позволяет полагаться на эффективную инфраструктуру Apache Spark для распределенного хранения и обработки графов.

Описание экспериментов

Системы и алгоритмы. Выбор систем и алгоритмов для экспериментов осуществлялся по двум критериям:

-     доступность исходного кода или интерфейса для программирования;

-     популярность системы в среде разработчиков и пользователей алгоритмов обработки графов.

На их основе были выбраны следующие системы:

-     GraphChi – наиболее популярная система для обработки графов на одном компьютере, используемая для сравнения в большом количестве статей, исходный код которой доступен на github;

-     PySpark, реализованная на основе Apache Spark; первоначально в качестве еще одной системы для сравнения рассматривалась GraphX, но оказалось, что возможностей PySpark достаточно для демонстрации особенностей вычислительных систем такого типа для выбранного алгоритма, при этом сложность реализации существенно ниже.

В экспериментах использован алгоритм INFOPATH [1], в котором для получения оценок изменяющейся во времени структуры и временной динамики предполагаемой сети применяется стохастический градиент [15]. При каждом запуске теста выполнялось 100 итераций, 10 000 случайно выбранных каскадов обрабатывалось на каждой итерации.

Выбор наборов данных. Входные данные для алгоритма INFOPATH представляют собой набор каскадов, описывающих распространение информации между узлами сети во времени. Каскад состоит из упорядоченной последовательности времен проявления информации и идентификаторов вершин производной длины, например: [(t1, v1), …, (tk, vk)], где k – длина каскада.

Для экспериментов использованы исходный набор данных для INFOPATH и синтетически сгенерированные данные (табл. 2).

Таблица 2

Характеристики наборов данных

Table 2

Datasets properties

Реализация алгоритма для системы GraphChi выполнена на языке программирования C++. Для оценки производительности использовалась система со следующими характеристиками: AMD Ryzen 7 3800X, 32 GB оперативной памяти, 1Tb NVMe диск. Результаты экспериментов представлены в таблице 3.

В процессе выполнения экспериментов обнаружены несколько особенностей вычислительной модели. GraphChi использует буфер ограниченного размера для хранения обновлений структуры графа на каждой итерации, соответственно, ограничено количество новых ребер, которые алгоритм может создать в тече- ние одной итерации. Этап переразбиения данных после некоторого числа изменений структуры графа требует более 100 тысяч открытых файлов, что может стать ограничением для отдельных систем. Надо отметить, что загрузка центрального процессора на разных стадиях обработки различается, а это ведет к неоптимальному использованию ресурсов.

Таблица 3

Результаты вычислений на основе GraphChi

Table 3

GraphChi-based computation results

Реализация алгоритма INFOPATH для системы Apache Spark с помощью PySpark выполнена на языке программирования Python, таким образом, результаты нельзя напрямую сравнить с реализацией GraphChi. Однако для системы на основе Apache Spark можно оценить масштабируемость на несколько узлов.

Для оценок производительности использованы кластеры Amazon EMR со следующими узлами: m5.2xlarge с 8 vcpu, 32 GB оперативной памяти, 128 GB EBS диск. Во всех экспериментах применялся набор данных INFOPATH internet memes.

Эксперименты показали, что реализация алгоритма Apache Spark хорошо масштабируется. Из графика на рисунке 4 видно, что на восьми узлах вычисления происходят в полтора раза быстрее (2 643 сек.), чем на четырех узлах (4 064 сек.).

Заключение

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

-     быстрые (GraphChi, TurboGraph) со статическим разбиением данных графа на разделы либо с отдельным этапом переразбиения при существенных изменениях;

-     в среднем более медленные (X-Stream, GraphChi DB), способные эффективно обрабатывать большие объемы изменений;

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

Такое разделение систем подтверждается полученными экспериментальными результатами для GraphChi и Apache Spark.

Анализ существующих решений показал, что проблема эффективного хранения и обработки динамических графов в полной мере не решена и требует дополнительного исследования.

Литература

1.    Gomez-Rodriguez M., Leskovec J., Schölkopf B. Structure and dynamics of information pathways in online media. Proc. VI ACM Int. Conf. WSDM'13, 2013, pp. 23–32. DOI: 10.1145/2433396.2433402.

2.    Kyrola A., Blelloch G., Guestrin C. GraphChi: Large-scale graph computation on just a PC. Proc. X USENIX Symposium OSDI'12, 2012, pp. 31–46. DOI: 10.21236/ada603410.

3.    Han W.S., Lee S., Park K., Lee J.H., Kim M.S., Kim J., Yu H. TurboGraph: a fast parallel graph engine handling billion-scale graphs in a single PC. Proc. XIX ACM SIGKDD Int. Conf. KDD'13, 2013, pp. 77–85. DOI: 10.1145/2487575.2487581.

4.    Cheng J., Liu Q., Li Z., Fan W., Lui J.C., He C. VENUS: Vertex-centric streamlined graph computation on a single PC. Proc. IEEE 31st Int. Conf. on Data Engineering, 2015, pp. 1131–1142. DOI: 10.1109/ICDE. 2015.7113362.

5.    Chen R., Shi J., Chen Y., Zang B., Guan H., Chen H. PowerLyra: Differentiated graph computation and partitioning on skewed graphs. Proc. X EuroSys, 2019, vol. 5, no. 3, pp. 1–39. DOI: 10.1145/2741948. 2741970.

6.    Jouili S., Vansteenberghe V. An empirical comparison of graph databases. Proc. Int. Conf. on Social Computing, 2013, pp. 708–715. DOI: 10.1109/SocialCom.2013.106.

7.    Kyrola A., Guestrin C. GraphChi-DB: Simple design for a scalable graph database system – on just a PC. ArXiv, 2014, art. 1403.0701. URL: https://arxiv.org/pdf/1403.0701.pdf (дата обращения: 27.09.2021).

8.    Aggarwal A., Vitter J. The input/output complexity of sorting and related problems. Communications of the ACM, 1988, vol. 31, no. 9, pp. 1116–1127. DOI: 10.1145/48529.48535.

9.    Roy A., Mihailovic I., Zwaenepoel W. X-Stream: edge-centric graph processing using streaming partitions. Proc. XXIV ACM SOSP, 2013, pp. 472–488. DOI: 10.1145/2517349.2522740.

10. Pissanetzky S. Sparse Matrix Technology. United States, Elsevier Science Publ., 1984, 336 p. DOI: 10.1016/c2013-0-11311-6.

11. Malewicz G., Austern M.H., Bik A.J. et al. Pregel: a system for large-scale graph processing. Proc. ACM SIGMOD Int. Conf. on Management of Data, 2010, pp. 135–146. DOI: 10.1145/1807167.1807184.

12. Apache Giraph. URL: https://giraph.apache.org/ (дата обращения: 27.09.2021).

13. Gonzalez J.E., Xin R.S., Dave A., Crankshaw D., Franklin M.J., Stoica I. Graphx: Graph processing in a distributed dataflow framework. Proc. XI USENIX Symposium OSDI, 2014, pp. 599–613.

14. Apache Spark. Unified Engine for Large-Scale Data Analytics. URL: https://spark.apache.org/ (дата обращения: 27.09.2021).

15. Agarwal A., Duchi J.C. Distributed delayed stochastic optimization. Proc. XXIV Int. Conf. NIPS, 2011, pp. 873–881. DOI: 10.1109/CDC.2012.6426626.

References

1. Gomez-Rodriguez M., Leskovec J., Schölkopf B. Structure and dynamics of information pathways in online media. Proc. VI ACM Int. Conf. WSDM'13, 2013, pp. 23–32. DOI: 10.1145/2433396.2433402.
2. Kyrola A., Blelloch G., Guestrin C. GraphChi: Large-scale graph computation on just a PC. Proc. X USENIX Symposium OSDI'12, 2012, pp. 31–46. DOI: 10.21236/ada603410.
3. Han W.S., Lee S., Park K., Lee J.H., Kim M.S., Kim J., Yu H. TurboGraph: a fast parallel graph engine handling billion-scale graphs in a single PC. Proc. XIX ACM SIGKDD Int. Conf. KDD'13, 2013, pp. 77–85. DOI: 10.1145/2487575.2487581.
4. Cheng J., Liu Q., Li Z., Fan W., Lui J.C., He C. VENUS: Vertex-centric streamlined graph computation on a single PC. Proc. IEEE 31st Int. Conf. on Data Engineering, 2015, pp. 1131–1142. DOI: 10.1109/ICDE.2015.7113362.
5. Chen R., Shi J., Chen Y., Zang B., Guan H., Chen H. PowerLyra: Differentiated graph computation and partitioning on skewed graphs. Proc. X EuroSys, 2019, vol. 5, no. 3, pp. 1–39. DOI: 10.1145/2741948.2741970.
6. Jouili S., Vansteenberghe V. An empirical comparison of graph databases. Proc. Int. Conf. on Social Computing, 2013, pp. 708–715. DOI: 10.1109/SocialCom.2013.106.
7. Kyrola A., Guestrin C. GraphChi-DB: Simple design for a scalable graph database system – on just a PC. ArXiv, 2014, art. 1403.0701. Available at: https://arxiv.org/pdf/1403.0701.pdf (accessed September 27, 2021).
8. Aggarwal A., Vitter J. The input/output complexity of sorting and related problems. Communications of the ACM, 1988, vol. 31, no. 9, pp. 1116–1127. DOI: 10.1145/48529.48535.
9. Roy A., Mihailovic I., Zwaenepoel W. X-Stream: edge-centric graph processing using streaming partitions. Proc. XXIV ACM SOSP, 2013, pp. 472–488. DOI: 10.1145/2517349.2522740.
10. Pissanetzky S. Sparse Matrix Technology. United States, Elsevier Science Publ., 1984, 336 p. DOI: 10.1016/c2013-0-11311-6.
11. Malewicz G., Austern M.H., Bik A.J. et al. Pregel: a system for large-scale graph processing. Proc. ACM SIGMOD Int. Conf. on Management of Data, 2010, pp. 135–146. DOI: 10.1145/1807167.1807184.
12. Apache Giraph. Available at: https://giraph.apache.org/ (accessed September 27, 2021).
13. Gonzalez J.E., Xin R.S., Dave A., Crankshaw D., Franklin M.J., Stoica I. Graphx: Graph processing in a distributed dataflow framework. Proc. XI USENIX Symposium OSDI, 2014, pp. 599–613.
14. Apache Spark. Unified Engine for Large-Scale Data Analytics. Available at: https://spark.apache.org/ (accessed September 27, 2021).
15. Agarwal A., Duchi J.C. Distributed delayed stochastic optimization. Proc. XXIV Int. Conf. NIPS, 2011, pp. 873–881. DOI: 10.1109/CDC.2012.6426626.