Near-Data Source Graph Partitioning

Abstract

Recently, numerous graph partitioning approaches have been proposed to distribute a big graph to machines in a cluster for distributed computing. Due to heavy communication overhead, these graph partitioning approaches always suffered from long ingress times. Also, heavy communication overhead not only limits the scalability of distributed graph-parallel computing platforms but also reduces the overall performance of clusters. In order to address this problem, this work proposed a near-data source parallel graph partitioning approach noted as NDGP. In NDGP, an edge was preferentially distributed to the machine where it was stored. We implemented NDGP over two classic graph partitioning approaches, Random and Greedy, and one most recently proposed graph partitioning approach, OLPGP, and evaluated its effectiveness. Extensive experiments conducted on real-world data sets verified the effectiveness of NDGP on reducing the communication overhead in the graph partitioning process and demonstrated that NDGP does not induce additional communication and computing workload to the graph-distributed computing that follows.

1. Introduction

Recently, graphs have been used to represent data coming from social science, astronomy, computational biology, telecommunications, semantic web, protein networks, and many other domains [1]. Also, the volume and complexity of these graph-structured data are increasing at a high rate. Processing these graph-structured data in effective way can help people to find knowledge with huge commercial or scientific value [2,3,4,5]. Graph partitioning can quickly split a big graph into small graphs for computing in a distributed way. Also, well-designed graph partitioning approaches are highly effective in minimizing the overall communication overhead and runtime of jobs with computing dependencies [6]. However, big data computing systems using existing graph partitioning strategies are still suffering from high communication overhead and low speedup. Thus, how to partition a big graph in an appropriate way with high efficiency becomes a significant research topic.

Graph partitioning methods vary significantly in terms of their heuristics, assumptions, and respective performance. Thus, this makes it difficult for developers to compare them and assess their characteristics. Choosing the right technique for the computational problem at hand is not trivial because each method adopts a limited set of application objectives and constraints.

Theoretically, graph partitioning is about cutting a large graph into several pieces. Partitioning a big graph into k pieces is usually called k-way partitioning. Some existing partitioning methods, e.g., Random partitioning, usually use a “one-size-fits-all” design that uniformly processes all vertices. Other existing partitioning methods, e.g., Greedy partitioning, try to find the best partition for a given graph under a certain objective function. So far, no graph partitioning method has paid attention to using the original location information of an edge to reduce the graph partitioning latency and the communication overhead introduced by graph partitioning. Due to the low network transmission rate among machines and the large amount of communication overhead, the graph partitioning process always suffers from long latency. In order to address this problem, this work proposed a new graph partitioning approach that takes the original location of an edge as a heuristic condition during the graph partitioning process. By reducing the communication overhead in the graph partitioning process, the proposed approach not only improves the graph partitioning rate but also relieves the overall communication overhead for the underlying system. Many partitioning techniques [4,7,8,9,10,11,12] work for data clustering, which is used for recommendations instead of workload partitioning. Work [13] used a graph partitioning approach for computational offloading in cloud computing.

The rest of this paper is organized as follows. Section 2 introduces the related works. The proposed near-data graph partitioning approach is detailed in Section 3 and Section 4. Section 5 details the experiment design and results. Section 6 concludes this paper.

2. Related Works

In terms of graph partitioning, there have been lots of work carried out in high-performance scientific simulation and multilevel graph partitioning during the recent decade [14,15,16,17,18,19]. Researchers have placed their focus on various subjects, such as dealing with large real-world graphs, large scale k-way balanced graph partitioning, and graph-parallel systems. Latest works have also explored 3D partitioning for large-scale graph processing [20].

To obtain adequate partitions in large real-world graphs, massive scale graphs frameworks, such as KaHIP [21], Metis [22], Scotch [23], are proposed. To obtain quick partitioning rate, parallel graph partitioning is also developed by people. Work [24] explores the design space of creating a multi-threaded graph partitioner. The idea of the parallel version of Scotch, PT-Scotch [23], is concerned about recursive bi-partitioning. Only the vertices near the border of a recent partitioning would be considered to reduce the communication cost in PT-Scotch. KaPPa [25] and PDiBaP [26] are also popular methods that are used in different situations, depending on the processors that are available. Compared to PT-Scotch, work [27] can achieve better effectiveness and scalability. While partitioning large complicated networks, ParHIP [27] works better as it adapts a label propagation technique.

Works [28,29] take the direction of an edge as a heuristic condition while partitioning the input graph. Some other methods, like Leopard [30], are proposed to process workloads that are read-only. However, it does not fit systems like Pregel.

In recent studies, Zhang et al. [31] introduced degree-based graph entropy in the similarity matrix calculation to realize social graph anonymization via a graph partition based on a privacy-preserving scheme, named GPPS. Shuai Zhang et al. [32] proposed a novel efficient and balanced vertex-cut graph partition algorithm called EBV, which grants appropriate weights to the overall communication cost and communication balance, and it had great potential in improving the huge communication volume or the extreme message imbalance among partitioned subgraphs in traditional graph partitioning algorithms. Moussawi et al. [33] compared the Spinner algorithm with their graph partitioning algorithm named B-GRAP, which was based on the Label Propagation (LP) approach and defined different objective functions to deal with either vertex or edge balance constraints while considering edge direction in graphs to show the improved performance of large graph data analytic computations. Additionally, Yin et al. [34] used a novel network partitioning algorithm to improve the scalability of agent-based modeling of mass evacuation based on a cutting-edge cyberGIS-enabled computational framework, applying graph partition theory successfully to situations like mass emergency evacuation. Moreover, LENTP (large-scale emulation network topology partition) based on the community detection with the weight of the vertex similarity has been proposed by Yan et al. [35] to solve the large-scale topology partition, like emulation networks. Angelika Wiegele and Shudian Zhao [36] introduced two graph partitioning problems, the k-equipartition problem and the graph partition problems with knapsack constraints (GPKCs), and used an extended ADMM algorithm, applying it to the tight SDP relaxations for graph partitioning problems with non-negativity constraints.

Many recent graph partitioning works have also adopted the re-partitioning strategy, in which the graph is dynamically redistributed across machines to further improve the load balance and reduce the communication overhead for the graph computing that follows [37,38,39,40].

3. Near-Data Graph Partitioning

In the big data era, the volume of data to process is always extremely huge. Utilizing distributed processing for these data is essential. Transmitting data among computing units obviously induces large volume communication overhead to the distributed computing system. Heavy communication overhead will introduce a long time latency to a big data processing job. Thus, instead of transmitting data to a computing unit, we think about moving the computing to data and trying to keep the data stay where they are. In large-scale distributed graph computing jobs, the task of graph partitioning is to assign the large graph to computing nodes. We employed the moving computing to data idea in a graph partitioning process, proposing the near-data graph partitioning (NDGP) approach. To be specific, when selecting a computing node for an edge, the node where the edge located is given priority under the premise that other basic requirements (e.g., work balance) are satisfied. Figure 1 illustrates the traditional graph partitioning process. Figure 2 illustrates the scenario under the near-data graph partitioning (NDGP) approach. We take Edge (1, 3) located in Machine 1 as an example. Assuming that assigning Edge (1, 3) to Machine 1, where it is located, does not violate the workload balance and other principles followed by traditional graph partitioning strategies, NDGP will assign Edge (1, 3) to Machine 1 for computing. On the other hand, traditional graph partition strategies will assign Edge (1, 3) randomly to a number of candidates. By keeping the local edge in the local machine, the cost of communication overhead due to assigning the edge to other machines is avoided. Thus, by introducing NDGP, the communication overhead occurring in the graph partitioning process is expected to be reduced. Consequently, the graph partitioning time is also expected to be shorter.

An effective distributed graph computing abstraction relies on an optimal graph placement on the computing nodes used. A successful graph partitioning approach should generate a graph placement that minimizes the computing and communication in the graph-distributed computing that follows and ensures work balance. In particular, given a graph $G(V,E)$, we assume that after p-way vertex-cut partitioning, the local graph assigned to a machine $M_{id}(id\in \{1,2,\dots ,p\}$ (p is the number of machines used)) is $G_{id}(V_{id},E_{id})$. Thus, in order to ensure work balance, the following conduction should be satisfied:

$$\underset{id\in \{1,2...p\}}{max}\left\{\right|E_{id}\left|\right\}\le \delta {\displaystyle \frac{\left|E\right|}{p}}$$

(1)

where $\delta (\delta \ge 1)$ is the balance factor that is a constant, which is set by system managers or users.

In order to minimize computing, I/O, and communication, we must minimize the number of vertex replicas. P-way vertex-cut partitioning assigns each edge $e\in E$ to a machine $\Omega \left(e\right)\in \{1,2,\dots ,p\}$. We note the set of machines a vertex v spans as $\Omega \left(v\right)\subseteq \{1,\dots ,p\}$. The balanced vertex-cut objective is

$$\underset{\Omega }{max}{\displaystyle \frac{1}{\left|V\right|}}\sum _{v\in V}|\Omega \left(v\right)|$$

(2)

$$s.t.\underset{id\in \{1,2\dots p\}}{max}\left\{\right|E_{id}\left|\right\}\le \delta {\displaystyle \frac{\left|E\right|}{p}}$$

(3)

4. Implementation

In this paper, we mainly discuss two existing representative graph partitioning approaches, Random [41] and Greedy [41]. Moreover, we implement and evaluate NDGP approaches over these two approaches. There are two reason for selecting Random and Greedy approaches to implement and evaluate NDGP. First, they are both basic and classic graph partitioning strategies. Other existing graph partitioning strategies are designed and implemented based on them, like NDGP. Second, NDGP is the first strategy that takes the data source into account in the graph partitioning process.

The Random approach simply conducts a vertex cut by assigning edges randomly to machines. It employs a hash map to ensure work balance. Thus, in order to design and implement the add-on near-data graph partitioning functionality, i.e., NDGP_Random (Algorithm 1), we just need to give high priority to the machine candidate that the current edge used for the assigning task is located on.

$$ks.t.arg\underset{k}{min}\mathbb{E}\left[\sum _{v\in V}|\Omega \left(v\right)|:{\Omega }_i,\Omega \left(e_{i+1}\right)=k\right]$$

(4)

Thus, in order to design and implement the near-data graph partitioning feature over the Greedy approach, i.e., NDGP_Greedy (Algorithm 2), the k-selected machine should be the machine where $e_{i+1}$ is located, and k must satisfy the following:

$$s.t.\underset{id\in [1,\dots p]}{max}\left\{\right|E_{id}\left|\right\}\le \delta {\displaystyle \frac{\left|E\right|}{p}}$$

(5)

$$s.t.arg\underset{k}{min}\mathbb{E}\left[\sum _{v\in V}|\Omega \left(v\right)|:{\Omega }_i,\Omega \left(e_{i+1}\right)=k\right]$$

(6)

Algorithm 1 NDGP_Random

// assign edge (source, target) to a machine p in {0,…numprocs-1} // β is imbalance factor INPUT: Source, Target, # of processes OUTPUT: process/machine ID 1: i = location_edge(source, target); 2: balance = (maxedges − proc_num_edges[i])/(epsilon + maxedges − minedges); 3: if balance $\le \beta$ then 4:     Return p = i; 5: end if 6: Return p = hash_edge(source, target) in {0, …numprocs-1};

Algorithm 2 NDGP_Greedy

// assign edge (source, target) to a machine p in {0,…numprocs-1} // β is imbalance factor INPUT: Source, Target, # of processes OUTPUT: process/machine ID   1: i = location_edge(source, target);   2: balance = (maxedges − proc_num_edges[i])/(epsilon + maxedges − minedges);   3: proc_score[i] = balance + a credit if source is already on i+ a credit if target is already on i;   4: if balance ≤ $\beta$ $\&\&$ (|proc_score[i] − maxscore < le-c|) then // c is an integer constant   5:     Return p = i;   6: end if   7: for each machine i do   8: balance = (maxedges − proc_num_edges[i])/(epsilon + maxedges − minedges);   9: proc_score[i] = balance + a credit if source is already on i + a credit if target is already on i; 10: if $\left|proc\_score\left[i\right]-maxscore\right|$< 1e-c then // c is an integer constant 11:     Put i into the candidate set; 12: end if 13: end for 14: Return p = hash_edge(source, target) in the candidate set;

The NDGP algorithm was implemented in the function assign_edge (source, target) provided by the open source distributed graph processing framework Powergraph [41]. NDGP works when the function assign_edge (source, target) is called. The time complexity of the NDGP algorithm is O(n), which is same to that of the existing graph partitioning algorithms, Random and Greedy. NDGP can reduce the communication overhead occurring in the graph partitioning process. In order to evaluate the performance of NDGP on the most recently proposed partitioning strategy, we also implemented NDGP on OLPGP [42]. OLPGP is an optimized label propagation-based graph partitiong algorithm implemented on Spark GraphX. We implemented NDGP_OLPGP by giving higher priority to the local machine if it was in the candidate set obtained by OLPGP.

5. Experiments

5.1. Experiment Platform

Our experiments were conducted on a Linux-based cluster. The cluster consists of one front end node that runs the Intel Data Center Manager and 140 computing (worker) nodes connected via 0.5 GB Ethernet. Each computing node has 64 GB of RAM and two E5-2690 CPUs (2.9 GHz/8-core). The job scheduler used is the IBM platform LSF job scheduling system. All data have three replicas that are stored in different storage nodes. We used up to 48 nodes in our experiment. The results and discussion may be presented separately, or in one combined section, and may optionally be divided into headed subsections.

5.2. Benchmarking Algorithms and Data Set

We selected PageRank, SSSP (directed shortest path) and Triangle Counting detailed in

Table 1. Summary of benchmarking algorithms.

Algorithm	Characteristics	Application Illustration	Category
PageRank	Iterative, high communication	Importance ranking	Direct Acyclic Graph
SSSP (shortest path)	Iterative, medium communication	Decision making	Direct Acyclic Graph
Triangle Counting	Single step, medium communication	Clustering coefficient	General

as benchmarking algorithms. These algorithms are all representative graph computing algorithms. Google uses the PageRank algorithm to compute the rank value for each web page in a web graph. This rank value determines the order of the web page in the show list when people search a page using Google. SSSP (directed shortest path) helps people to find the shortest path between two nodes in a graph. Triangle Counting is mainly used to evaluate the cohesion of a specific network and the degree of aggregation of nodes. It can be used for social network analysis and fraud analysis. These algorithms are used as benchmarks and are implemented by most existing open source distributed graph processing frameworks. We conducted our tests on Spark GraphX [43] and Powergraph [41]. We evaluated NDGP on four data sets listed in

Table 2. Summary of data sets.

Graph	Description	Type	# of Vertices	# of Edges	Graph Density ($\times {10}^{-5}$)	Average Degree	Memory Size (GB)
soc-LiveJournal	Friednship social network	Directed	4,847,571	68,993,773	0.59	14	10.3
Twitter	Social networks	Directed	41,652,230	1,468,365,182	0.08	35	231.2
BFS	Facebook social networks	Directed	61,876,615	336,776,269	0.02	5	51.6
com-Orkut [44]	Orkut online social network	Undirected	3,072,441	1,171,850,837	2.48	76	22.1

.

5.3. Experiment Design and Results

We conducted parallel graph partitioning and ran the PageRank algorithm on the selected data sets to evaluate the effectiveness of NDGP. All presented results came from the average of at least three runs. In our experiments, edge distributing was conducted on all computing nodes in parallel. Thus, NDGP was conducted in parallel. By distributing the edges in parallel, the graph partitioning time can be significantly reduced compared to the sequential graph partitioning using only one computing node.

Figure 3 compares the number of edges sent when we are distributing the LiveJournal data set to eight machines using the Greedy partitioning strategy and the NDGP_Greedy partitioning strategy, respectively. Figure 4 compares the number of edges sent when we are distributing the Twitter data set to 16 machines using the Random partitioning strategy and the NDGP_Random partitioning strategy, respectively. The thickness of the line represents the number of edges. The results demonstrate that compared to the original partitioning strategies, NDGP can significantly reduce the number of edges sent. For example, compared to the Random partitioning strategy, the NDGP_Random partitioning strategy can reduce the number of edges sent up to 33%.

Figure 5 and Figure 6 compare the runtime and the volume of communication overheads happening on the graph computing process running PageRank on the Twitter data set, respectively. Both the runtime and the volume of communication overheads under NDGP and the existing strategies are roughly the same. This demonstrates that NDGP does not lengthen the runtime and not introduce more communication overheads while running PageRank on the Twitter graph comparing the existing strategies. Moreover, NDGP does not affect the effectiveness of Greedy and OLPGP in reducing the runtime and communication overheads of PageRank running on the Twitter graph. The reason is that NDGP does not introduce an additional operation that will affect the graph computing that follows, with the number of machines used increasing the results under NDGP and existing strategies being constantly almost the same. This verifies the scalability of NDGP and also demonstrates that NDGP does not break the workload balance obtained by existing partitioning strategies.

Figure 7 shows a comparison of the volume of communication overheads happening on the graph computing process running SSSP on the BFS data set, and Figure 8 shows a comparison of the graph computing time following the partitioning phase running Triangle Counting on the BFS data set. The results demonstrate that NDGP introduces no additional computing and communication overheads to SSSP and Triangle Counting. The results also demonstrate that NDGP does not affect the effectiveness of the Greedy partitioning strategy in reducing the SSSP and Triangle Counting runtime.

METIS [45] uses a technique called multi-level graph construction, which first refines and balances large undirected graphs multiple times and then performs k-way partitioning at each layer. K-way partitioning divides a graph into k parts with equal or approximately equal number of edges. This algorithm optimizes the partitioning quality by minimizing edge cutting (i.e., the number of edges connecting different partitions). In order to fully demonstrate the effectiveness of NDGP, we also integrated NDGP into the METIS partitioning strategy and conducted comparative experiments. Due to METIS being used to deal with undirected graphs [46], we conducted our experiments on an undirected graph data set, com-Orkut. Figure 9 shows a comparison of the number of edges sent when we were distributing the com-Orkut data set. The results demonstrate that compared to METIS, NDGP_METIS can also significantly reduce the number of edges sent, which dramatically reduce the communication overhead in the graph partitioning progress. Figure 10 and Figure 11 demonstrate that NDGP does not induce a negative impact and even further improves the efficiency of METIS, in some cases, regarding both the communication overhead and runtime while calculating the partitioned graph.

In summary, NDGP can significantly reduce the graph partitioning communication overhead and time by processing local edges with the local machine. On the other hand, NDGP introduces no additional computing and communication overhead to the graph computing process and does not affect the effectiveness of the Greedy partitioning strategy in reducing the graph computing time. Also, the scalability of the NDGP approach was demonstrated on both the size of the data sets and the number of machines used. Good scalability guarantees the practical applications of the proposed approach.

6. Conclusions

Due to the low network transmission rate among machines and large amount of communication overhead, the graph partitioning process always suffers from long latency. In order to address this problem, we proposed a near-data source parallel graph partitioning approach, NDGP. NDGP takes the original location of an edge as a heuristic condition during the graph partitioning process. We designed and implemented the proposed approach and evaluated its effectiveness on real-world data sets. Our extensive experimental results verified the effectiveness of our approach on reducing the communication overhead for the graph partitioning process and demonstrated that the proposed approach did not induce additional communicating and computing workload to the following graph-distributed computing. Also, the scalability guaranteeing the practical applications of the proposed approach was verified on both the size of the data sets and the number of machines used. In summary, the proposed approach reduced the graph partitioning overhead and improved the effectiveness of the distributed graph processing system. In the future, we plan to take the edge location information into account in the graph computing phase, aiming to reduce the communication overhead and runtime in the graph computing process.