Graphs provide a powerful abstraction for representing real world networked data [1]. Examples such as social networks [2], biological networks [3] and transportation networks [4] illustrate their ubiquity. Extreme scale graph analytics is imperative for several real-world Big Data applications with the underlying graph structure containing millions or billions of vertices and edges [5].
This work is sponsored by the U.S. NSF under grant numbers OAC-1911229, PPoSS-2119816 , SaTC-2104264.
The massive sizes of graphs have made distributed processing essential. Local dedicated clusters are available to the researchers and practitioners at several institutions to perform distributed graph processing. In addition, public clouds, with their flexibility, elasticity and pay-as-you-go-models are also available for utilization. Moreover, the availability of several distributed graph processing frameworks, which allow easier distributed application development, has further increased the adoption of distributed graph processing [6].
In a typical distributed graph processing framework [7]- [9], the graph is partitioned and assigned to various compute nodes. The connectivity across the partitions induces communication among the compute nodes. Communication latency is significant in distributed clusters/clouds. To reduce communication, most distributed frameworks focus on choosing the right computation models or partitioning scheme.
While such design choices do reduce communication, they exert a second order impact on the same. The communication scheme, which we define as the choice of data (vertex attributes, intermediate updates, etc.) to be communicated and the virtual interconnection network among the communication nodes, has a first-order impact on the communication performance. Formal analysis on the impact of communication schemes is lacking. A brute force approach wherein an application developer evaluates all possible configurations of number of nodes and communication schemes can be prohibitively expensive both computationally and economically. With increasing adoption of cloud platforms for graph processing, such an analysis is imperative for addressing the challenges due to high cloud latencies.
In this work, we explicitly decouple the communication and computation in order to estimate the impact of communication in distributed environments. We develop performance estimation models for communication schemes used by existing most popular distributed graph processing frameworks. The performance models take as input the partitions of the graph and the communication scheme and output the estimated communication time. Thus, our estimation models enable a preliminary analyses of the trade-offs between the partitioning schemes and the communication in an early development stage. We validate our models using a local university cluster and the cloud hosted NSF Chameleon cluster [10].
The significant contributions of this work are as follows:
• We identify the communication schemes used by the most popular distributed graph processing frameworks; • Given partitions of a graph, we develop models that estimate the performance of these communication schemes; • We validate our performance estimation models with experiments on a local and a cloud hosted cluster; • We use our models to provide several crucial insights:
-We provide a concrete explanation behind the superior performance of frameworks that perform edge proportional communication (Section V-C-Insight 3). -We show that even though edge proportional communication frameworks have lower communication in absolute terms, the scalability of frameworks performing vertex proportional communication is much superior (Section V-C-Insight 2). -We show a hypothetical scenario where such vertex proportional communication frameworks can outperform the edge proportional communication frameworks in terms of communication time (Section V-C-Insight 4). -We conclude that partitioning scheme has no impact on vertex proportional communication. -We also formulate the requirements of the optimal partitioning scheme for edge proportional communication frameworks and show that graph partitioning heuristics should reduce the number of vertices which have at least one incoming edge from other partitions as opposed to minimizing the number of edges in the cut-set that is typically done (Section V-C-Insight 5).
In this work, we focus on distributed in-memory graph processing frameworks which use message passing for communication [6]. Examples of such frameworks include Pregel [11], Giraph [12], iGiraph [8], etc. In line with the aforementioned frameworks, factors such as the time for fetching data from disks into main memories are treated as preprocessing steps and not considered in the analysis.
Platform Definition: We assume a distributed graph analytics platform consisting of a cluster of N compute nodes for processing a graph G = (V, E). We assume that the data are represented at the granularity of words, which is platform dependent. To model the communication time of a message, we let t s denote the average communication latency between two nodes, i.e., the time taken for the destination node to receive the first word from the source node over the network. Let t w denote the average time it takes to transfer a word. Finally, let d denote the average degree of G.
Distributed Graph Processing Abstraction: We consider the Graph Processing Over Partitioning (GPOP) [13] abstraction for distributed graph processing. In this abstraction the graph G is partitioned into several partitions. The partitions are assigned to the N -node cluster. In each iteration, each node performs computations over the partitions assigned to it. The results of the computation are communicated using a partition-centric approach, i.e., the communication occurs at the granularity of partitions as opposed to vertices or edges. This process continues for a fixed number of iterations or until the algorithm converges. Furthermore, as the communication between the partitions assigned to the same node occurs via local memory and does not induce a communication on the network, the discussion in this paper, henceforth, centers around nodes of the cluster. Widely used frameworks such as Pregel, Giraph, iGiraph, etc. can be easily modeled using this abstraction.
Analysis of Existing Frameworks: In [14], authors develop a centralized framework with a master node dispatching the tasks and collecting the results (updated vertex attributes). In [15], the authors developed a reduce-scatter algorithm based on a ring logic. Each node computes a portion of the entire results (vertex attributes), and repeatedly delivers a portion to its successive neighbor on the logical ring. Frameworks such as Pregel [11], Giraph [12], iGiraph [8], Gluon [16] and those developed in [17], [18], etc. compute a unique message for each destination vertex using the source vertex attribute and the edge attribute. Thus, the number of messages generated is proportional to the number of (incoming) edges in the graph.
Communication Scheme Definition: By analyzing the most popular graph processing frameworks, we identify the dimensions along which they differ with respect to their communication characteristics.
I. Type of Data Being Communicated: A framework performs either of the following: (i) Broadcast the vertex attributes in the communication phase. In the computation phase, each node selects the appropriate broadcast vertex attributes using the list of incoming edges of the vertices in its partitions. (ii) In computation phase, each node generates a unique message for every outgoing edges of the vertices in its partitions. In communication phase, all-to-all personalized communication of the message is performed. We denote the former scheme as Vertex Proportional Communication (VPC) while the latter as Edge Proportional Communication (EPC).
II. Underlying Virtual Communication Network: A framework either uses a centralized scheme where a master node collects all the data and sends it to the worker nodes or it uses a distributed network, where the nodes communicate directly with each other. We denote the centralized network as Master-Worker (MW). For distributed network based frameworks, frameworks that use EPC perform all-toall personalized communication with direct communication between nodes. We denote this network as Peer-to-Peer (P2P). However, for frameworks that use VPC and require a broadcast operation, [15] shows that an embedded ring logic is an efficient network for the same. We denote this network as Ring.
Based on the analysis in the previous section, we now formally define the communication schemes.
1) Vertex Proportional Communication (VPC): We say that a distributed application follows Vertex Proportional Communication scheme if the messages being communicated are the vertex attributes. Note that we do not make any assumptions regarding the computational model. Algorithm 1 shows an example of VPC-based PageRank algorithm. Each vertex computes its PageRank value and then broadcasts the value to other vertices so that they can use it in the next iteration. Thus, the amount of data to communicate depends on the number of vertices in the graph.
Input:
Now, based on the virtual interconnection network, we further categorize VPC into two categories: a) Vertex Proportional Communication on Master-Worker Network (VPC-MW): In this category, the distributed application follows a master-worker model. As shown in Figure 1(a), every worker node is connected to a master node. There is no direct data communication between worker processes. All-to-all Broadcast is implemented in two steps. In the first step, each worker node in the cluster sends its data to the master, and the master node collects data from all worker processes. In the second step, the master node broadcasts the data received in the first step to every worker.
b) Vertex Proportional Communication on Ring Network (VPC-Ring): In this category, the distributed application uses a virtual ring network to implement broadcast operation. Allto-all Broadcast: Let us denote the N nodes as M 0 , M 1 , where k = 1, 2, . . . , N . In Iteration 1, each node M k sends its initial data to its right neighbor M (k+1)%N , and receives data from its left neighbor M (k+N -1)%N . Both sending and receiving operations are non-blocking primitives, so they can be performed at the same time. After receiving data from its left neighbor, each node stores data in its local memory. For the remaining iteration j (j = 1, 2, ..., N -1), each node sends data that was initially contained in M (k-j+1)%N to its right neighbor M (k+1)%N , and receives data that was initially contained in M (k-j)%N from its left neighbor M (k+N -1)%N . After (N -1) iterations, every node can get a copy of data originated from all other nodes. [15] also proves that such implementation on ring network is bandwidth optimal.
2) Edge Proportional Communication (EPC): We say that a distributed application follows Edge Proportional Communication scheme if the messages being communicated are generated due to an operation on the edge weight and the attribute of the source vertex. Here again, we do not make any assumptions regarding the computational model. The virtual interconnection network that we assume in this case is a fully connected Peer-to-Peer network, i.e., each node can communicate with every other node directly. We do not append P2P to EPC as this is the only virtual network that is used. Algorithm 2 shows an example of EPC-based PageRank. Each vertex computes its contribution to each of its neighbors. The computed values are communicated among nodes using all-to-all personalized communication. We also consider a widely used optimization [7], [16] that combines messages destined to the same vertex from a partition.
Remark: [6] provides a classification of distributed processing frameworks based on graph topology. Under that terminology, our categorization is applicable to the frameworks falling under the Vertex-Centric (Edge-Cut) and Component-Centric frameworks. Also note that there is no one-to-one correspondence between their categorization and ours, i.e., a Vertex-Centric (Edge-Cut) (or a Component-Centric framework) can follow either VPC or EPC scheme.
In this section, we build models to estimate the performance of the aforementioned communication schemes. The models take the partitions, their logical mapping to nodes and the communication scheme as input.
Let V denote the number of vertices and N denote the number of nodes. If multiple partitions are mapped to a node, we consider the union of them as a single partition. Thus, the number of partitions is equal to the number of nodes in our model. For VPC-MW scheme, there exists an additional node acting as master besides worker nodes. The average communication latency between two nodes is denoted by t s while t w denotes the per word transfer time in the communication network.
(1) For VPC-MW scheme, the communication time is
Assuming that each worker receives data from the master sequentially, the first term is the time for sending a vector with a length of V , containing attributes for all vertices, from the master to workers. The second term is the time for sending a vector with a length of V /N from every worker to the master.
(2) For VPC-Ring scheme, the execution time for performing a broadcast on the ring can be estimated as
In each step, every node performs two actions: receiving a vector with a length of V /N from its left neighbor and sending a vector with the same length to its right neighbor. All the nodes take the two actions at the same time. Such steps are repeated (N -1) times in a broadcast process.
(3) For EPC scheme, the communication time is given as
Here, α ij denotes the number of vertices in partition j that have at least one incoming edge from partition i. Formally, α ij = A ji 1 0 , where A ji denotes the sub-matrix in the adjacency matrix of the graph with rows corresponding to partition j and columns corresponding to partition i and 1 denotes all ones vector of suitable size. η ij denotes the average size of the combined messages. For commutative operator, η ij = 1. Note that in the worst case, α ij = V N . Given a graph partitioning, we can calculate the parameters V , N and α ij . However, to obtain the values of t s and t w , measurements on the target platform are required which is discussed in the next subsection.
We perform the following experiment on two platforms (more details in Section V-A) -HPC platform and Chameleon Cloud to estimate the value of latency t s and throughput t w .
On each platform, we use MPI primitives to perform round-trip communication between two different nodes and measure the communication time. Let L denote the length of communicated data. The round-trip time can be estimated as T RT T = 2(t s +L×t w ). We can estimate t s and t w by varying the value of L and fitting the measured T RT T using a linear function with the least mean square error. For the Chameleon cloud, we estimate the latency and throughput as 0.35 ms and 1.05 ms/MiB, respectively. For HPC, we estimate them to be 0.2 ms and 0.20 ms/MiB, respectively.
For a thorough design space exploration, an application developer will need to vary the number of nodes, partitioning techniques, communication patterns, etc. leading to a combinatorial explosion of the design space. Moreover, given the increasing large scale of graphs -billions of vertices and edges, a sub-optimal choices will have tremendous time and monetary costs as typical graph processing algorithms run for hundreds or thousands of iterations. Our performance estimation model can enable quick trade-off analysis. Additionally, our model can be used to understand the impact of various parameters, e.g., communication schemes, number of nodes, on the performance of graph processing applications. Conclusions drawn from the estimations can help accelerate design space exploration. For example, as per Section V-C-Insight 5, partitioning strategy has no impact on VPC communication time with VPC models and need not be analyzed.
We conduct our experiments on two computing platforms. First, we experiment on the High-Performance Cluster (HPC) provided by the university. We use machines with Dual Intel We choose PageRank and Weakly Connected Components (WCC) algorithms as our benchmarks. We perform experiments using uk-union-2006-06-2007-05, twitter-2010 and webbase-2001 each containing over 1 billion edges with average degree varying from 8.63 to 41.21 (Table I). The graphs, processed by Webgraph [21] and LLP [22] tools, have very high locality due to optimized vertex ordering.
We implement each algorithm using the three communication schemes. For each algorithm and communication scheme, we conduct experiments with {16, 32, 64, 128, 256} partitions mapped to {1, 2, 4, 8, 16} nodes respectively. That is each node, with 16 processor cores, is assigned 16 partitions so that each partition can be processed in parallel. For Chameleon, we failed to allocate enough nodes for the case of 256 partitions. As a pre-processing step, we first partition the initial graphs into subgraphs and save them into the file system. For each experiment, we measure communication time per iteration by averaging over all the iterations of the algorithm. Figures 2345show the results. Dashed lines represent the estimations derived from our performance model while solid lines represent experimental results. As the platforms are running other applications at the same time, communication congestion occurs depending upon the network utilization in the data center. The impact of such congestion is hard to predict since it does not depend upon the applications we analyze. However, it is still evident that the solid lines are close to corresponding dashed lines or have similar trends in the graphs, thereby validating our models.
A few observations from the graphs are as follows: VPC-MW has a worse performance than VPC-Ring and EPC. VPC-MW's execution time increases almost linearly and hence it has a poor scalability compared with VPC-Ring. EPC performs better than VPC-Ring experimentally.
Insight 1: VPC-Ring and EPC consistently outperform VPC-MW.
We derive this conclusion from both our model and experimental results. Since the latency t s is relatively much lower than communication time, we can simplify our model by ignoring terms with t s . The estimated communication time for the three schemes are as follows
From the simplified model we can see that VPC-MW requires O(N ) times higher communication that VPC-Ring. Moreover, even in the worst case the communication of EPC is slightly better than VPC-MW. Insight 2: VPC-Ring has the best scalability. Both the simplified model and experimental results show that the communication time for VPC-Ring remains constant with the increasing number of partitions. Whereas, the communication time for VPC-MW and EPC increases almost linearly. For the latter two schemes, there is a trade-off between the communication time and the storage at each machine, which is determined by the number of partitions. Using VPC-MW or EPC as communication scheme, if we divide the original graph into more partitions, more time will be spent on communication, but each machine will require less storage. Insight 3: In practice, EPC outperforms VPC-Ring.
For a partition
po , where d i po is the out degree of partition i, i.e., total number of edges with source vertex in partition i but destination vertex not in it. Now,
V , where d po is the average out degree of the partitions. In real-world graphs which follow power law, a partitioning that leads to high in-partition connectivity and low connectivity across partitions will lead to a very small value of d po . Thus, N dpo V < 1 implies better performance for EPC. Insight 4: Hypothetical scenario where VPC-Ring will outperform EPC.
The above analysis indicates that VPC-Ring will outperform EPC when
This happens if the partitioned graph has low locality and few vertices from the same node share common destinations, which increases the average out degree among the partitions. Insight 5: Impact of partitioning on the performance of the communication schemes.
As per our models, both T V P C-ring and T V P C-mw are only dependent on the number of vertices in the graph and the number of partitions/nodes. Thus, partition scheme has no impact on the communication performance of VPC frameworks. An application choosing this scheme can focus on partitioning which is optimal for computation only.
For EPC, assuming η ij = 1, an optimal partitioning is given by min N i=1 j =i α ij . As α ij denotes the number of vertices in partition j that have at least one incoming edge from partition i, an optimal partitioning will minimize the number of such vertices across all pairs of partitions. Clearly, this is an NP-hard problem due to the l 0 norm. Heuristics can be developed to explicitly optimize for this objective. Note that a partitioning that reduces the number of edges in the cut-set is given by j =i A ji 1 1 , i.e., l 1 norm instead of l 0 . Thus, instead of focusing on minimizing the l 1 norm, heuristics should explicitly focus on minimizing the l 0 norm.
The main focus of this paper is to model communication performance of graph processing frameworks rather than developing a new framework. Thus, we limit the discussion in this section on works that either focus on communication optimizations or develop performance models for communication time estimation for graph processing frameworks.
Communication Focused Frameworks: McCune et al. [18] provided a detailed survey of message passing and shared memory based graph processing. Heidari et al. [6] characterized communication models into message passing, shared memory and push/pull style. Both works characterize the communication models but do not provide a quantitative analysis of the communication performance. Patarasuk et al. [15] presented a framework that uses virtual ring interconnect (Ring in our work) to optimize bandwidth utilization for clusters with tree topology. Alfatafta et al. [23] proposed algorithms and architectures for several MPI collective operations, including using ring scheme to implement AllReduce and AllGather operations. However, it relies on the knowledge of the underlying system architecture and network structure which might not always be available.
Communication Performance Modeling of Graph Processing Frameworks: Xu et. al. [24] developed a performance model for vertex centric graph algorithms that captures communication bandwidth and communication latency. However, their target platform is a shared memory single node architecture and the communication is defined as the data transfer between the LLC (Last Level Cache) and external memory. In our work, we focus on modeling inter-node communication within a cluster/cloud platform. Al-Tawil et. at. [25] used LogGP, which is a simple performance model that reflects the most important parameters required to estimate the communication performance of parallel computers, to evaluate the performance of message passing interface (MPI) on different cluster platforms. Their main focus was on modeling at MPI primitive level whereas we model at a higher abstraction level of communication schemes as defined in this paper. Abdolrashidi et. al. [26] developed a model for computation and communication for vertex-centric graph algorithms and used the model to explore the trade-off of different partitioning schemes. Thus, their model can estimate communication performance for only vertex-centric algorithms. In contrast, by focusing on how the data is communicated instead of how it is generated, our model is agnostic to the underlying computation model. Thus, our model is applicable to a wide range of graph processing frameworks. Bhimani et. al. [27], [28] developed queuing theory based performance prediction model for distributed MPI systems. Their communication model is similar to the Master Worker Model (Section III-B) which requires one-to-many and many-to-one node communication.
In contrast, we also model P2P communication model as widely used graph processing frameworks employ this model. Moreover, their focus is to perform accurate performance prediction of a given application, thereby requiring complex queuing theory and Machine Learning based modeling of packet transfer costs. In contrast, the focus of this paper is to develop a model that can enable trade-off analyses of various partitioning and communication scheme choices.
We will focus on the following: Improving Performance Estimations: Our focus in this work was to enable early stage trade-off analysis of various design choices. While our models already capture these trade-offs and trends in general, there is still some room to improve the accuracy for each individual configuration. Similar to [28], we plan to develop stochastic and Machine Learning (ML) based models for network latency and round trip time predictions; and Modeling Non-Stationary Algorithms: In this work, we focused on stationary algorithms, i.e., algorithms in which all vertices participate in computations in all the iterations [13]. In contrast, in non-stationary algorithms, the vertices which participate in computations varies between iterations, e.g., Shortest path problems [13]. We will focus on developing heuristics to estimate average per-iteration performance.
In this work, we developed and validated performance estimation for communication schemes used in popular distributed graph processing frameworks. The models enable analyses of the trade-offs between the partitioning schemes and the communication schemes in early development stages.
Current distributed graph processing frameworks have done extensive research on improving the performance by optimizations such as partitioning and load balancing. However, with increasing adoption of cloud platforms, optimizations which drastically reduce the communication time will be required. We believe a formal analysis into the communication schemes as presented in this paper will accelerate such innovations.