Most real-world networks evolve over time. This evolution can be modeled as a series of graphs that represent a network at di erent points in time. Our G* system enables e cient storage and querying of these graph snapshots by taking advantage of the commonalities among them. We are extending G* for highly scalable and robust operation. This paper shows that the classic challenges of data distribution and replication are imbued with renewed signi cance given continuously generated graph snapshots. Our data distribution technique adjusts the set of worker servers for storing each graph snapshot in a manner optimized for popular queries. Our data replication approach maintains each snapshot replica on a di erent number of workers, making available the most e cient replica con gurations for di erent types of queries.
Real-world networks, including social networks and the
Web, constantly evolve over time [
Several single-graph systems are available today: Google's
Pregel [
Single-graph systems typically distribute the entirety of a single graph over all workers to maximize the bene ts of parallelism. When there are multiple graph snapshots, however, distributing each snapshot on all workers may slow down query execution. In particular, if multiple snapshots are usually queried together, it is more advantageous to store each snapshot on fewer workers as long as the overall queried data are balanced over all workers. In this way, the system can reduce network overhead (i.e., improve query speed) while bene ting from high degrees of parallelism. We present a technique that automatically adjusts the number of workers in a manner optimized for popular queries.
As implied above, there are vast di erences in execution time depending on the distribution con gurations and the number of snapshots queried together. Replication gives us, in addition to enhanced system reliability, the opportunity to utilize as many distribution con gurations as there are replicas. G* constructs r replicas for each snapshot to tolerate up to r 1 simultaneous worker failures. Our technique classi es queries into r categories and optimizes the distribution of each replica for one of the query categories.
In this paper, we make the following contributions: We de ne the problem of distributing graph snapshots and present a solution that expedites queries by adjusting the set of workers for storing each snapshot. We provide a technique for adaptively determining replica placement to improve system performance and reliability.
We present preliminary evaluation results that show the e ectiveness of the above techniques.
We discuss our research plans to complete the construction of a highly scalable and reliable system for managing large graph snapshots.
The remainder of the paper is organized as follows: Section 2 presents the research context and provides formal de nitions of the problems studied in the paper. Sections 3 and 4 describe our new techniques for distributing and replicating graph snapshots. Section 5 presents our preliminary evaluation results. Section 6 discusses related work. Section 7 concludes this paper.
α 2.1 (3/4, {G1}), (4/5, {G2}), (5/6, {G3}) aveavragge (1, 2, {G1, G2, G3}), (1, 1, {G1, G2, G3}), (2, 0, {G1}), (3, 1, {G2} (4, 2, {G3}) union (1, 2, {G1,G2,G3}) (1, 1, {G1,G2,G3}) count_,sum count_sum (a, 2, {G1,G2,G3}) (b, 1, {G1,G2,G3}) degree degree (a, ..., {G1,G2,G3}) (b, ..., {G1,G2,G3}) vertex vertex a c c c e (2, 0, {G1}), (3, 1, {G2}), (4, 2, {G3})) count_sum (c, 0, {G1}), (d, 0, {G1, G2}), (c, 1, {G2, G3}), ... degree (c, ..., {G1}), (d, ..., {G1, G2}), (c, ...,{G2, G3}), ...
vertex b {G1,G2,G3} β
b d {G1,G2,G3} γ
d d f {G1} {G1,G2} {G2,G3} {G3}
G* is a distributed system for managing large graph
snapshots that represent an evolving network at di erent points
in time [
To track all of these vertex versions, each worker
maintains a Compact Graph Index (CGI) that maps each
combination of vertex ID and graph ID onto the disk location
that stores the corresponding vertex version. For each vertex
version (e.g., c2), the CGI stores only one (vertex ID, disk
location) pair in a collection for the combination of
snapshots that contain that vertex version (e.g., fG2; G3g). In
this manner, the CGI handles only vertex IDs and disk
locations while all of the vertex and edge attributes are stored
on disk. Therefore, the CGI can be kept fully or mostly in
memory, enabling fast lookups and updates. To prevent the
CGI from becoming overburdened by managing too many
snapshot combinations, each worker automatically groups
snapshots and then separately indexes each group of
snapshots [
Query Processing. Like traditional database systems, G* supports sophisticated queries using a data ow approach where operators process data in parallel. To quickly process queries on multiple graph snapshots, however, G* supports special operators that share computations across snapshots.
CGI c1
c2 c disk {G1} {G2,+G3} c c e {G1,+G2} {G3} d d f e1 e e d1
d2 d
f1 f f
Accelerating computation by distributing data over
multiple servers has been a popular approach in parallel
databases [
Example. Consider a scenario where each of 100 similarlysized graph snapshots contains approximately 1 million vertices and 100 million edges. Assume also that the system consists of one master and 100 workers. Table 1 compares two snapshot distribution con gurations: Shared-Nothing, where each of the 100 snapshots is stored on one distinct worker, and Shared-Everything, where each snapshot is evenly distributed over all of the 100 workers. For each of these con gurations, two types of queries for computing the PageRank of each vertex are executed: Query One Snapshot, and Query All Snapshots. The explanations below are based on our evaluation results (see Section 5 for details).
In the case of Shared-Nothing, querying one snapshot using only one worker takes 285 seconds (205 seconds to construct the snapshot from disk and 80 seconds to run 20 iterations of PageRank). Querying all snapshots on all workers in parallel takes the same amount of time. When the SharedEverything con guration is used, querying one snapshot on all workers takes approximately 22 seconds, mainly due to network communications for the edges that cross worker boundaries (the disk I/O and CPU costs correspond to only 205/100 seconds and 80/100 seconds, respectively, due to the distribution of the snapshot over 100 workers). In this con guration, querying 100 snapshots takes 2,205 seconds as the PageRank of each vertex varies across graph snapshots, thereby causing 100 times more message transmissions than the previous case. This example shows the bene ts of different snapshot distribution approaches for di erent types of queries (e.g., Shared-Nothing for queries on all snapshots and Shared-Everything for queries on one snapshot). Formal De nition. Our ultimate goal is to keep track of the popularity of graph snapshots and to optimize the storage/distribution of unpopular snapshots for space e ciency (Section 2.1) and popular snapshots for query speed. In this paper, we focus on the problem of distributing popular snapshots over workers in a manner that minimizes the execution time of queries on these snapshots. This problem can be formally de ned as follows:
Problem 1. (Snapshot Distribution) Given a series of graph snapshots fGi(Vi; Ei) : i = 1; 2; g, n workers, and a set of queries Q on some or all of the snapshots, nd a distribution fVi;w : i = 1; 2; ^ w = 1; 2; ; ng that minimizes Pq2Q time(q; fVi;wg) where Vi;w denotes the set of vertices that are from snapshot Gi(Vi; Ei) and that are assigned to worker w, and time(q; fVi;wg) represents the execution time of query q 2 Q on the distributed snapshots fVi;wg satisfying (1) [nw=1Vi;w = Vi (i.e., the parts of a snapshot on all workers cover the original snapshot) and (2) Vi;w \ Vi;w0 = ; if w 6= w0 (i.e., workers are assigned disjoint parts of a snapshot).
Our solution to the above problem is presented in Section 3. 2.2.2
Snapshot Replication
There have been various techniques for replicating data to
improve availability and access speed [
Problem 2. (Snapshot Replication) Given a series of graph snapshots fGi(Vi; Ei) : i = 1; 2; g, the degree of replication r, n workers, and a set of queries Q on some G2,1 G2,2 " G3,1 G3,2 ! G2,1 G2,2 " G1,1 G1,2 (a) Before Exchange (b) After Exchange or all of the snapshots, nd a replica distribution fVi;j;w : i = 1; 2; ^ j = 1; 2; ; r ^ w = 1; 2; ; ng that minimizes Pq2Q time(q; fVi;j;wg) where Vi;j;w denotes the set of vertices that are from the jth replica Gi;j (Vi;j ; Ei;j ) of snapshot Gi(Vi; Ei) and that are assigned to worker w, and time(q; fVi;j;wg) denotes the execution time of query q on the distributed snapshot replicas fVi;j;wg satisfying (1) [nw=1Vi;j;w = Vi;j = Vi for j = 1; 2; ; r, (i.e., the parts of a snapshot replica on all workers cover the original replica), (2) Vi;j;w \ Vi;j;w0 = ; if w 6= w0 (i.e., workers are assigned disjoint parts of a snapshot replica), and (3) Vi;j;w \Vi;j0;w = ; if j 6= j0 (i.e., no worker w contains multiple copies of a vertex and its edges, which tolerates r 1 simultaneous worker failures).
Section 4 presents our solution to the above problem. 3.
As mentioned in Section 2.2.1, G* needs to store each
graph snapshot on an appropriate number of workers while
balancing the utilization of network and CPU resources.
In contrast to traditional methods for partitioning a static
graph [
Our dynamic data distribution approach meets the above requirements. In this approach, each G* worker partitions its graph data into segments with a certain maximum size (e.g., 10GB) so that it can control its load by migrating some segments to other workers (Section 3.1). Our approach continuously routes incoming messages for updating vertices and edges to appropriate workers with low latency (Section 3.2). When a segment becomes full, G* splits that segment into two that are similar in size while maintaining data locality by keeping data accessed together within the same segment (Section 3.3). It does all of the above while supporting G*'s graph processing operators (Section 3.4). 3.1
In G*, each worker periodically communicates with a randomly chosen worker to balance graph data. Our key principles in load balancing are to (1) maximize the bene ts of parallelism by uniformly distributing data that are queried together and (2) minimize network overhead by co-locating data from the same snapshot. Consider Figure 3(a) where three snapshots are partitioned into a total of 6 similarlysized segments. In this example, each of workers and are assigned a segment from snapshot G1, is assigned two segments from G2, and is assigned two segments from G3. If snapshots G1 and G2 are frequently queried together (see those shaded in Figure 3(a)), this snapshot distribution leads to ine cient query execution due to imbalanced workload between the workers and network communications for the edges between G1;1 and G1;2. This problem can be remedied by exchanging G1;1 and G3;1 between the workers, which results in a balanced distribution of the data queried together (i.e., G1 and G2) and localized processing of G1 on and G2 on , respectively.
Given a pair of workers, our technique estimates, for each segment, the bene t of migrating that segment to the other worker, and then performs the most bene cial migration. This process is repeated a maximum number of times or until the migration bene t falls below a prede ned threshold. The bene t of migrating a segment is calculated by multiplying the probability that the segment is queried with the expected reduction in query time (i.e., the di erence between expected query time before and after migration).
For a set Si of segments on worker i and another set Sj of segments on worker j, the expected query time is computed as Pq2Qk p(q) time(q; Si; Sj) where Qk is a collection of k popular query patterns, p(q) is the probability that query pattern q is executed, and time(q; Si; Sj) denotes the estimated duration of q given segment placements Si and Sj.
Our technique obtains Qk (equivalently, k popular
combinations of segments queried together) as follows: Sort
segments from Si [ Sj in order of decreasing popularity.
Initialize Qk (for storing k popular query patterns) with the
rst segment. Then, for each of the remaining segments,
combine it with each element from Qk and insert the result
back into Qk. Whenever jQkj > k, remove its least popular
element. We estimate the popularity of each combination
of segments by consolidating the counting synopses [
We compute time(q; Si; Sj) as max(c(q; Si); c(q; Sj)) + c0(q; Si; Sj) where c(q; Si) is the estimated duration of processing the segments from Si for query q, and c0(q; Si; Sj) represents the estimated time for exchanging messages between workers i and j for query q. 3.2
Each new vertex (or any edge that emanates from the vertex) is rst routed to a worker chosen according to the hash value of the vertex ID. That worker assigns such a vertex to one of its data segments while saving the (vertex ID, segment ID) pair in an index similar to the CGI (Section 2.1). If a worker receives an edge that emanates from an existing vertex v, it assigns that edge to the segment that contains v. If a worker w has created a segment S and then migrated it to another worker w0 for load balancing reasons (Section 3.1), worker w forwards the data bound to S to w0. To support such data forwarding, each worker keeps track of the worker location of each data segment that it has created before. Updates of vertices and edges, including changes in their attribute values, are handled as in the case of edge additions. This assignment of graph data to workers is scalable because it distributes the overhead of managing data over workers. It also proceeds in a parallel, pipelined fashion without any blocking operations. 3.3
If the size of a data segment reaches the maximum (e.g.,
10GB), the worker that manages the segment creates a new
segment and then moves a half of the data from the
previous segment to the new segment. To minimize the number
of edges that cross segment boundaries, we use a traditional
graph partitioning method [
G*'s graph processing operators, such as those for
computing clustering coe cients, PageRank, or the shortest
distance between vertices, are usually instantiated on every
worker that stores relevant graph data [
G* masks up to r 1 simultaneous worker failures by
creating r copies of each graph data segment. As discussed in
Sections 2.2 and 3, the optimal distribution of each graph
snapshot over workers may vary with the number of
snapshots frequently queried together. Based on this
observation, we developed a new data replication technique that
speeds up queries by con guring the storage of replicas to
bene t di erent categories of queries. This approach uses
an online clustering algorithm [
Updates of vertices and edges are handled as described in Section 3.2 except that they are routed to r data segment replicas on di erent workers. For this reason, each worker keeps a mapping that associates each segment ID with the r workers that store a replica of the segment. Our approach protects this mapping on worker w by replicating it on workers (w+1)%n; (w+2)%n; ; (w+r 1)%n where n denotes the number of workers. If a worker fails, the master assigns another worker to take over. 4.2
The replicas of a data segment are split in the same way due to the use of a deterministic partition method. For each vertex migrated from one data segment to another, the r workers that keep track of that vertex update their (vertex ID, segment ID) pairs accordingly. 4.3
For each query, the master identi es the worker locations of the data segment replicas to process. To this end, the
Disk I/O bandwidth Snapshot construction in memory PageRank iteration per snapshot 1M messages/sec 200 Mbytes/sec 200 seconds 4 seconds master keeps track of the mapping between graph snapshots and the data segments that constitute them. The master also maintains the mapping between data segment replicas and the workers that store them. Using these mappings, the master selects one replica for each data segment such that the overall processing load is uniformly distributed over a large number of workers and the expected network overhead is low. Next, as Figure 1 shows, the master instantiates operators on these workers and starts executing the query. 4.4
Each worker balances its graph data as explained in Section 3. The only di erence is that whenever a query of category j accesses a replica of a data segment, the counting synopsis of the j-th replica of the data segment is updated using the ID of the query (Section 3.1). In this way, the j-th replica of each segment is assigned to a worker in a manner optimized for query category j.
This section presents our preliminary results obtained by running G* on a six-node, 48-core cluster. In this cluster, each machine has two Quad-Core Xeon E5430 2.67 GHz CPUs, 16GB RAM, and a 2TB hard drive. We plan to extend these experiments with more queries on larger data sets in a bigger cluster (Section 7).
To construct a realistic example in Section 2.2.1, we measured the overhead of key operations summarized in Table 2. In our evaluation, a worker was able to transmit up to 1 million messages to other workers within a second, although a 1Gbps connection may enable 10 million transmissions of 12-byte messages in theory. The reason behind this result is that there is inherent overhead when writing and creating message objects to and from TCP sockets in Java. Furthermore, reading approximately 1Gbytes of data from disk to construct a graph snapshot took 5 seconds. However, constructing a snapshot in memory by creating 100 million edge objects and registering them in an internal data structure took approximately 200 seconds.
In the next set of experiments, we created a series of 500 graph snapshots using a binary tree generator. Each snapshot in the series was constructed by rst cloning the previous snapshot and then inserting 20,000 additional vertices and edges to the new graph. Therefore, the last graph in the series contained 10 million vertices. We ran a query that computes, for each graph, the distribution of the shortest distances from the root to all other vertices. Table 3 shows, for the shortest distance query, the speedup achieved by distributing the snapshots over more workers. The highest speedup was achieved with 48 workers. This table also
80.5 seconds 19.2 seconds 53.2 seconds shows that the relative bene t of data distribution (i.e., the speedup relative to the number of workers) tends to decrease with more workers. This is mainly due to increased network tra c, which shows the importance of balancing CPU and network resources in the context of continuously creating large graph snapshots.
The e ectiveness of two di erent distributions is demonstrated in Table 4. If most queries access only the largest snapshot, then it is bene cial to distribute that snapshot over all workers to maximize query speed. On the other hand, if all of the snapshots are queried together, our approach stores each graph on a smaller subset of workers to reduce network overhead. In this case, all of the workers can still be used in parallel since the entire graph data is distributed over all workers. The bene ts of distribution con gurations are less pronounced in Table 4 than Table 1 due to a smaller number of message transmissions and fewer workers. Table 4 also demonstrates the bene t of G* in executing queries on multiple snapshots. In particular, the time for processing 500 snapshots (e.g., 80.5 seconds) is only up to 10 times longer than that for processing the largest snapshot (e.g., 8.2 seconds) since the computations on the largest snapshot are shared across smaller snapshots. 6.
In this section, we brie y summarize related research, focusing on previous graph systems, data distribution, and data replication.
Previous Graph Systems. In contrast to systems which
process one graph at a time [
Data Replication. There has been extensive work on data
replication that focused on improving data availability and
performance [
We presented G*, a scalable and robust system for storing and querying large graph snapshots. G* tackles new data distribution and replication challenges that arise in the context of continuously creating large graph snapshots. Our data distribution technique e ciently stores graph data on the y using multiple worker servers in parallel. This technique also gradually adjusts the number of workers that store each graph snapshot while balancing network and CPU overhead to maximize overall performance. Our data replication technique maintains each graph replica on a di erent number of workers, making available the most e cient storage con gurations for various combinations of queries.
We are working on full implementations of the techniques presented in this paper to enable new experiments with additional queries on larger data sets. We will analyze these techniques to classify their complexity. We plan to look into the challenges of scheduling groups of queries, dealing with varying degrees of parallelism, resource utilization, and user-generated performance preferences. We are exploring failure recovery techniques for long-running queries while exposing the tradeo between recovery speed and execution time. We also want to study opportunities for more granular splitting, merging, and exchanging of data at the vertex and edge level rather than in large segments as discussed in this paper. We intend to seek opportunities for gains in execution speed at the expense of storage space by segregating recent and popular \hot" data (which we could store in a less compressed manner) from less popular \cold" data (which could be highly compressed).