Anne-Marie Kermarrec

François Taïani

Juan M Tirado

Scaling Out Link Prediction with SNAPLE: 1 Billion Edges and Beyond

Keywords: Graphs, Link Prediction, Big Data, Distributed Systems * ASAP

In this paper, we consider how the emblematic problem of link-prediction can be implemented efficiently in gather-apply-scatter (GAS) platforms, a popular distributed graph-computation model. Our proposal, called SNAPLE, exploits a novel highly-localized vertex scoring technique, and minimizes the cost of data flow while maintaining prediction quality. When used within GraphLab, SNAPLE can scale to extremely large graphs that a standard implementation of link prediction on GraphLab cannot handle. More precisely, we show that SNAPLE can process a graph containing 1.4 billions edges on a 256 cores cluster in less than three minutes, with no penalty in the quality of predictions. This result corresponds to an over-linear speedup of 30 against a 20-core standalone machine running a non-distributed state-of-the-art solution.

Passage à l'échelle de la prédiction de liens dans les grands graphes avec SNAPLE: 1 milliards d'arètes et au delà Résumé : Dans cet article, nous considérons comment le problème emblématique de la prédiction de liens dans les grands graphes peut être mis en oeuvre de façon efficace sur les plates-formes du modèle rassembler-appliquer-diffuser (Gather Apply Scatter ou GAS), un nouveau modèle de calcul réparti particulièrement prometteur pour le traitement distribué des grands graphes. Notre proposition, appelée SNAPLE, exploite une nouvelle technique de notation de sommets particulièrement localisée. Cette forte localité lui permet de minimiser le coût des échanges de données, et donc les temps de calcul, tout en maintenant une forte qualité des liens prédits. Lorsqu'elle est utilisée au dessus de GraphLab (un moteur de référence du modèle GAS), SNAPLE peut traiter de très grands graphes, allant bien au delà des limites d'un algorithme classique de prédiction de liens réalisé directement sur GraphLab. Plus précisément, nous montrons que SNAPLE est capable de prédire les liens d'un graphe contenant 1,4 milliards d'arètes sur un cluster de 256 coeurs en moins de trois minutes, sans pénaliser la qualité des prévisions obtenues. Ce résultat correspond à un rapport d'accélération sur-linéaire de 30 comparé à la performance d'une solution centralisée de l'état de l'art tournant sur une seule machine de 20 coeurs.

Mots-clés : Graphes, Prédiction de Liens, Big Data, Systèmes Distribues Link prediction [START_REF] Liben-Nowell | The link prediction problem for social networks[END_REF] seeks to predict missing edges from a graph G, with direct application to recommendation, data mining or search. Edges might be missing because the graph is evolving over time (users create new social links), or because G only captures a part of a wider ground truth. Predicted edges can be used to recommend new users (social graphs), new items (bipartite graph), or uncover missing information (social mining). Formally, link prediction considers two graphs G = (V, E) and G = (V, E ) so that G contains less information than G in the form of missing edges: E E . For instance G and G might represent the same social graph captured at different points in time, or G might represent an incomplete snapshot of a larger graph represented by G (an interaction network, a set of related topics, etc.). The goal of link prediction is then to predict which are the edges of G that G lacks, i.e. to determine E \ E.

Link prediction strategies fall into unsupervised and supervised approaches. The typical approach for unsupervised link prediction is sketched in Algorithm 1. The algorithm executes a parallelizable loop that iterates through all vertices of G (lines 1-3). Each iteration hinges on the scoring function score(u, z) (line 2) which reflects how likely the edge (u, z) is to appear in G . In its basic version, the algorithm scores all vertices z that are not already in u's neighborhood (noted Γ G (u) = {v ∈ V |(u, v) ∈ E}), and returns the k vertices with the highest scores as predicted neighbors for u (operator argtop k ).

The function score(u, z) may only use topological properties, such as the connectivity-based metrics proposed in the seminal work of Liben-Nowell and Kleinberg [START_REF] Liben-Nowell | The link prediction problem for social networks[END_REF] (e.g. the number of common neighbors between u and z, |Γ G (u) ∩ Γ G (z)|). This score may also exploit vertex content, i.e. the additional application-dependent knowledge attached to vertices [START_REF] Brzozowski | Who should i follow? recommending people in directed social networks[END_REF][START_REF] Schifanella | Folks in folksonomies: social link prediction from shared metadata[END_REF][START_REF] Rowe | Who will follow whom? exploiting semantics for link prediction in attention-information networks[END_REF][START_REF] Parimi | Predicting friendship links in social networks using a topic modeling approach[END_REF], such as user profiles, tags, or documents. In many domains, pure topological metrics tend to be the main drivers of link generation, and are therefore almost always present in the prediction process. Scoring functions using only topological metrics are also more generic as they do not rely on information out of the graph (tags, content, user profile, etc.) that might not be available in all data sets. Supervised approaches build upon unsupervised strategies and leverage machine-learning algorithms to produced optimized scoring functions [START_REF] Scellato | Exploiting place features in link prediction on locationbased social networks[END_REF][START_REF] Lichtenwalter | New perspectives and methods in link prediction[END_REF]. Supervised approaches tend to perform better, but at the cost of an important learning effort, as they must often scan the whole graph to build an accurate classification model. In this paper, we therefore focus on unsupervised approaches, but the key ideas we present can be extended to supervised schemes.

Scaling link prediction

While research on link prediction originally sought to maximize the quality of predicted edges, with little consideration for computation costs, its practical relevance for social networks and recommendation services has put it at the forefront of current system research. One critical challenge faced by current implementations is the fast growing size of the graphs they must process [START_REF] Gupta | Wtf: the who to follow service at twitter[END_REF][START_REF] Lempel | The stochastic approach for link-structure analysis (salsa) and the tkc effect[END_REF].

A first strategy to scale Algorithm 1 focuses on the performance of individual iterations. A frequent optimization limits the search for missing edges to the vicinity of individual vertices. If one notes Γ K G (u)

Inria the K-hop neighborhood of u, defined recursively as

Γ 1 G (u) = Γ G (u) Γ K G (u) = {z ∈ V |∃v ∈ Γ K-1 G (u) : (v, z) ∈ E} (1)
this optimization will only consider the vertices of Γ K G (u) \ Γ G (u) at line 2 as potential new neighbors in G , instead of the much larger set V \ Γ G (u). K is generally small (2 or 3). We use K = 2 in this work.

prediction u ← argtop k z∈Γ K G (u)\Γ G (u) score(u, z) (2)
This optimization works well because social graphs, and field graphs in general, tend to present high clustering coefficients. As a result, most of the edges to be predicted in G will connect vertices only separated by a few hops in G [START_REF] Yin | Structural link analysis and prediction in microblogs[END_REF]. Other optimization on standalone machines leverage specialized data structures and memory layout to exploits data locality and minimize computing costs [START_REF] Kyrola | Graphchi: Large-scale graph computation on just a pc[END_REF][START_REF] Roy | X-stream: Edge-centric graph processing using streaming partitions[END_REF].

A second strategy seeks to scale Algorithm 1 horizontally by deploying it to a distributed infrastructure. That is the case of Twitter for instance, who recently moved their Who-to-Follow service to a distributed solution, from an initial single machine deployment [START_REF] Gupta | Wtf: the who to follow service at twitter[END_REF]. This transition can leverage a growing number of graph processing engines [START_REF] Gupta | Wtf: the who to follow service at twitter[END_REF][START_REF] Bershad | The midway distributed shared memory system[END_REF][START_REF] Johnson | Crl: High-performance all-software distributed shared memory[END_REF][START_REF] Basumallik | Programming distributed memory sytems using openmp[END_REF][START_REF] Lumsdaine | Challenges in parallel graph processing[END_REF][START_REF] Malewicz | Pregel: A system for large-scale graph processing[END_REF][START_REF] Salihoglu | Gps: A graph processing system[END_REF][START_REF] Khayyat | Mizan: a system for dynamic load balancing in large-scale graph processing[END_REF], which aim to facilitate the realization of scalable graph processing tasks. These engines do so by implementing highly parallelizable programming models such as Bulk Synchronous Processing [START_REF] Valiant | A bridging model for parallel computation[END_REF][START_REF] Malewicz | Pregel: A system for large-scale graph processing[END_REF][START_REF] Salihoglu | Gps: A graph processing system[END_REF][START_REF] Khayyat | Mizan: a system for dynamic load balancing in large-scale graph processing[END_REF] (BSP), or Gather, Apply, Scatter (GAS) [START_REF] Low | Distributed graphlab: A framework for machine learning and data mining in the cloud[END_REF] .

The GAS model

In this work, we focus particularly on the GAS model, which can be seen as a refinement of map-reduce and BSP for graphs. Its reference implementation, GraphLab [START_REF] Low | Distributed graphlab: A framework for machine learning and data mining in the cloud[END_REF][START_REF] Gonzalez | Powergraph: Distributed graph-parallel computation on natural graphs[END_REF], is particularly scalable, and was found to perform best in a recent comparison of modern graph engines across a number of typical graph computing tasks [START_REF] Han | An experimental comparison of pregel-like graph processing systems[END_REF].

More precisely, a GAS program assumes every vertex u ∈ V and every edge (u, v) ∈ E of a graph G = (V, E) is associated with some mutable data, noted D u and D (u,v) . A GAS program consists of a sequence of GAS super-steps (steps for short), each comprising three conceptual phases that execute in parallel at each vertex and manipulate these data. (Our notation follows closely that of [START_REF] Gonzalez | Powergraph: Distributed graph-parallel computation on natural graphs[END_REF].) 1. The gather phase is similar to a map-and-reduce step. This phase collects the data associated with a vertex u's neighbors Γ G (u) and with u's outgoing edges {u} × Γ G (u). It maps this data through a user-provided gather() function; and aggregates the result using a general sum() function defined by the user.

Σ ← sum v∈Γ G (u) gather(D u , D (u,v) , D v ) (3)
2. In the apply phase, the result of the gather phase, Σ, is used to update the current data of node u, D u using a user-defined apply function.

D u ← apply Σ, D u (4)
3. Finally the scatter phase uses Σ and the new value of D u to update the data associated with u's outgoing edges u × Γ G (u) using a scatter function provided by the user.

∀v ∈ Γ G (u) : D (u,v) ← scatter Σ, D u , D (u,v) (5)
N 0 N 1 N 2 c : Γ G (c) d : Γ G (d) e : Γ G (e)
Figure 1: Example of data propagation in order to compute score(a, c), score(a, d) and score(a, e) following a naive GAS approach for a graph deployed into three different computing nodes (N 0 , N 1 and N 2 ). We have to propagate Γ G (c), Γ G (d) and Γ G (e) through b and f (dashed blue arrows). Then, this information has to be transferred to vertex a which is allocated into a different machine (N 1 ). Additionally, this information has to be replicated onto the replicas of a for synchronization.

Link prediction in the GAS model

The GAS model facilitates the scheduling and parallelization of vertex operations while increasing content locality. GAS engines, however, can require some substantial effort to adapt existing graph algorithms, for two reasons. First, graph traversals, a primitive strategy of many graph algorithms, are difficult to express in the GAS paradigm without adding substantial complexity and overhead. This is because the access and updates of a GAS step are limited to adjacent vertices and edges.

The second difficulty pertains to the limited access to topological information offered by the GAS paradigm. In the GAS model, vertices drive and organize the computation (principally in the gather and scatter phases), but are not expected to be an object of computation per se, in order to deal with densely connected vertices in power-law distributions. Vertices and neighborhoods are therefore not exposed directly as accessible data.

These two limitations are particularly problematic for the link prediction algorithm captured in equation [START_REF] Backstrom | Group formation in large social networks: Membership, growth, and evolution[END_REF]. This algorithm traverses all vertices in the K-hop neighborhood Γ K G (u) of individual vertices, and requires access to topological information in its 'score()' function (see Figure 1). Choosing a naive approach to work around these limitations, by propagating state, and replicating data across vertices, can be extremely counter-productive, and tend self-defeat the design rationale of the GAS paradigm.

The SNAPLE framework

To address the above challenges, we have developed a novel and lightweight framework that computes a similarity score score(u, z) between two vertices u and z without the costs associated with traditional similarity metrics. Our framework avoids a direct and costly computation of similarity by propagating intermediate results along the paths of the graph G. This propagation involves two low-overhead steps: in a first path-combination step we combine raw similarity values along 2-hop paths in G, resulting in a path-similarity for each 2-hop paths connecting u to z. In a second path-aggregation step, we then aggregate these path-similarity values to compute score(u, z). These two steps are configurable with userprovided functions for combination and aggregation, and provide in effect a novel scoring framework tailored to the GAS paradigm.

Path-combination

Our approach starts from a "raw" similarity metric, that we use as a basic building block to construct our scoring framework. We consider topological metrics that can be computed from the neighborhoods of the two nodes u and z one wishes to compare:

sim(u, z) = f Γ(u), Γ(z) (6)
where f is a similarity metric on sets, for instance Jaccard's coefficient [START_REF] Salton | Introduction to Modern Information Retrieval[END_REF]. This approach can be extended to content-based metrics [START_REF] Hasan | A survey of link prediction in social networks[END_REF] by simply including data attached to vertices in f . In the following, we limit ourselves to the 2-hop neighborhood of u when searching for candidate nodes, i.e. we use K = 2 in (2), a typical value for this parameter. As explained earlier, a first challenge when directly using the similarity shown above in the GAS model to implement the score(u, z) function of equation ( 2), is the inherent difficulty to access data attached to the nodes z ∈ Γ2 (u) \ Γ(u), which are not direct neighbors of u. One naive solution consists in using an initial GAS step to propagate a node's information to its neighbors, and make this data accessible to neighbors of neighbors in subsequent steps.

D u .neighborhood ← (v, D v ) v ∈ Γ(u) (7)
Unfortunately, and as we will show in our evaluation, the redundant data transfer and additional storage this approach causes make it highly inefficient, yielding counterproductive results in particular on very large graphs.

In order to overcome this limitation, SNAPLE uses a path-combination step that returns a similarity value (termed path-similarity) for each 2-hop path u → v → z connecting a source vertex u with a candidate vertex z:

sim v (u, z) = sim(u, v) ⊗ sim(v, z) (8)
where ⊗ is a binary operator that is monotonically increasing on its two parameters, such as a linear combination, or generalized means. We call the operator ⊗ a combinator 2 . Intuitively, equation ( 8) seeks to capture the homophily often observed in field graphs: if u is similar to v and v to z, then u is likely to be similar to z.

Table 1: Examples of combinators ⊗ name a ⊗ b sim v (u, z) linear αa + (1 -α)b αsim(u, v) + (1 -α)sim(v, z) eucl √ a 2 + b 2 sim(u, v) 2 + sim(v, z) 2 geom √ a × b sim(u, v) × sim(v, z) sum a+b sim(u, v) + sim(v, z) count 1 1
Table 1 lists five examples of combinators that we consider in this work: a linear combination, the Euclidean distance, the geometric mean, a plain sum (a special case of linear combination), and a basic counter (a degenerated case where all similarity values are stuck to 1).

Path-aggregation

The path-combination step we have just described provides a similarity value sim v (u, z) for each 2-hop path u → v → z connecting u to z. Multiple such paths are however likely to exist. For instance, in

u v x y z f sim(u, v) sim(v, x) sim(v, y) sim(v, z) sim(u, f ) sim( f , z) score(u, z) = w∈{v, f } sim w (u, z)
Figure 2: Two paths connect u to z, resp. through v and f . Each path yields different similarity values: sim v (u, z) and sim f (u, z). To summarize these different values, SNAPLE aggregates them using an aggregator (⊕, Table 2).

Figure 2, z can be reached from u over two 2-hop paths:

u → v → z and u → f → z, delivering two different similarity values, sim v (u, z) and sim f (u, z). SNAPLE aggregates these different values to obtain the final score of a node z ∈ Γ 2 (u) \ Γ(u): score(u, z) = v∈Γ(u)∩Γ -1 (z) sim v (u, z) (9)
where Γ -1 (u) is the inverse neighborhood of u, and ⊕ is a multiary operator that can be decomposed in a generalized sum ⊕ pre (the incremental application of a commutative and associative binary operator), and a normalization function ⊕ post that takes the results of ⊕ pre and its number of arguments as input to produce its result:

x∈X x = ⊕ post x∈X pre x , |X| (10)
We call the operator ⊕ an aggregator. This second step, termed path aggregation, is akin to a reduce step, and particularly well adapted to the GAS model. Several ⊕ operators can be used such as addition, multiplication, selecting the largest similarity, etc. We use three in this paper, Sum, arithmetic means (Mean), and geometric means (Geom), which we list in Table 2.

Table 2: Examples of aggregators ⊕ name a ⊕ pre b ⊕ post (σ , n) ⊕ x∈X x Sum a + b σ ∑ x∈X x Mean a + b 1 n σ 1 |X| ∑ x∈X x Geom a × b σ 1 n (∏ x∈X x) 1 |X|
Path-aggregation enhances the design of classic similarity metrics by adding information regarding the number of paths connecting a source and sink vertices. The combination of a similarity metric sim(), a combinator ⊗, and an aggregator ⊕ creates the design space of SNAPLE's similarity framework. In Table 3, we show an excerpt of the possible combinations we investigate in this work by combining elements of Tables 1 and2.

Inria Using Jaccard's coefficient on neighborhoods as our similarity metric, we systematically combine the first three combinators (⊕) of Table 1 (linear, eucl, and geom) with the three aggregators of Table 2 (Sum, Mean, Geom) to obtain a total of nine scoring mechanisms. By adapting the similarity and combinator, we are also able to define two extra scoring methods (gray rows) : a score function similar to personalized page rank [START_REF] Bahmani | Fast personalized pagerank on mapreduce[END_REF] (PPR), and a basic scoring approach (counter) that counts the number of 2-hop paths existing between u and z. This large set shows both the expressiveness and flexibility of the SNAPLE framework. Individual aggregators can greatly affect the final behavior of the resulting scoring function. Figure 3 illustrates this impact when using the linear combinator (α = 0.5) with the three aggregators of Table 2. In this example, a is connected to e and f through two 2-hop paths, and to g through three 2-hop paths. The highest score obtained with each aggregator is shown in bold in table.

The Mean and Geom aggregators (two bottom lines in the table), average out the number of paths connecting a to each candidate vertices, and as a result, consider f the vertex most similar to a. By contrast, the Sum aggregator takes into account the connectivity of candidate vertices in its final score, and rates g over f (first line). This means a vertex with lower path-similarities (such as g here) can obtain a high final score if enough paths connect it to the source vertex (a). Finally, let us note how the Geom aggregator penalizes vertices such as e which are connected through paths with very low path-similarity (here a → h → e).

Implementing SNAPLE in GAS

The approach we have just presented addresses the challenges inherent to the GAS model that we discussed in Section 2.4. Even using SNAPLE's vertex score however, the candidates space Γ 2 (u) \ Γ(u) of equation ( 2) remains generally too large to be fully explored. Indeed, if we note n the average out degree of vertices in G, a blind application of (2) can require up to O(n 2 ) scores to be computed per vertex on average, leading to O(|V |n 2 ) scoring operations for the whole graph G = (V, E).

This challenge is not inherent to the GAS model, but must be addressed in our implementation to obtain a tractable solution. We attack it using a two-pronged approach: first, we truncate neighborhoods that are larger than a truncation threshold thr Γ (with thr Γ reasonably large, e.g. 200), to limit the memory overhead induced by very large neighborhoods, and minimize the cost of computing raw similarity values. We note Γ(u) this truncated neighborhood, which we use to compute the raw similarity values of equation [START_REF] Bershad | The midway distributed shared memory system[END_REF].

Second, we do not consider all the paths u → v → z over Γ 2), where

Γ 2 (u) to Γ max k local 2 (u) in equation (
Γ max k local (u) = argtop k local v∈Γ(u) f Γ(u), Γ(z) (11)
and (Γ max k local ) 2 is defined in relation to Γ k local as explained in equation ( 1). The resulting GAS program is shown in Algorithm 2, and illustrated on a small example in Figure 4. The program comprises three GAS steps, each made of a gather (gather and sum functions), and an apply phase (apply function). We do no use any scatter phase. We employ the notation used in [START_REF] Low | Distributed graphlab: A framework for machine learning and data mining in the cloud[END_REF] where D u , D v and D (u,v) are the program state and meta-data for vertices u, v and edge (u, v) respectively.

In the first step (lines 1-6, Fig. 4b), we construct a list of neighbors by collecting the ids of adjacent vertices. To limit memory overheads, we limit the size of the neighborhood set D u . Γu to the truncation threshold thr Γ . This is achieved with a uniform random variable rand() over [0, 1] (line 3), yielding a good approximation of the truncation under GAS restrictions. Step 1: Obtain a sample of u's neighbors D u .

Γ 1: gather 1 (D u , D (u,v) , D v ): 2: γ ← {v} 3: if rand() > thr Γ /|Γ(u)| then γ ← / 0 ; 4: return γ 5: sum 1 (γ 1 ,γ 2 ): return γ 1 ∪ γ 2 Computing Σ 1 6: apply 1 (D u , Σ 1 ): D u . Γ ← Σ 1
Step 2: Estimate similarities

7: gather 2 (D u , D (u,v) , D v ): 8: return v, sim(u, v) ≡ f ( D u . Γ , D v . Γ ) 9: sum 2 (γ 1 ,γ 2 ): return γ 1 ∪ γ 2 Computing Σ 2 10: apply 2 (D u , Σ 2 ): 11: D u .sims ← argtop k local (v,s v )∈Σ 2 (s v )
Step 3: Compute recommendations In the second step (lines 7-11, Fig. 4c), we use the truncated neighborhoods D x . Γx to compute raw similarities between u and each of its neighbors v (line 8). At the end of this step, we only keep the top k local neighbors of u (argtop k local operator, line 11), and store them into a dictionary D u .sims with their corresponding raw similarity s v . When this second step terminates, the keys of D u .sims are the vertices of Γ max k local (u) as defined in [START_REF] Gonzalez | Powergraph: Distributed graph-parallel computation on natural graphs[END_REF].

In the final step (lines 12-20, Fig. 4d), we first compute path similarities (Sec. 3.1) using the combinator ⊗ (line 15). These path similarities are limited to the the k local 2-hop neighborhood of u, i.e. to paths u → v → z such that v ∈ D u .sims.keys (line 13) and z ∈ D v .sims.keys (line 15). We then aggregate the path-similarities leading to the same candidate vertex z as explained in Section 3.2 using the ⊕ pre (line 16) and ⊕ post operators (line 19). More precisely, γ 1 , γ 2 and Σ 3 are sets of triplets (z, s z , n z ) ∈ V × R × N that associate each candidate vertex z to a similarity s z and a counter n z . n z counts the number of paths over which s z has been accumulated. The function merge at line 16 performs a double fold (or reduce) operation on the vertices of γ 1 and γ 2 to compute the generalized sum and the count of equation [START_REF] Gonzalez | Graphx: Graph processing in a distributed dataflow framework[END_REF], which are then used at line 19:

merge(⊕ pre , γ 1 , γ 2 ) = z, pre (z,s z ,-) ∈γ 1 ∪γ 2 s z , ∑ (z,-,n z ) ∈γ 1 ∪γ 2 n z z ∈ γ 1↓V ∪ γ 2↓V
where γ i↓V is the projection of γ i on its first component V . The program finally returns the top k vertices with the best scores as predictions (line 20).

Evaluation

We present a detailed experimental evaluation of SNAPLE implemented on top of the Graphlab distributed graph engine [START_REF] Gonzalez | Powergraph: Distributed graph-parallel computation on natural graphs[END_REF]. Graphlab implements the GAS paradigm over an asynchronous distributed shared memory abstraction, and is specifically designed for processing large graphs in distributed deployments.

We explore the space of possible link predictors that can be designed using SNAPLE's scoring framework and the implementation presented in Algorithm 2. In particular, we evaluate how different configurations modify the quality of link predictions, and the total computing time.

Experimental setup

We have implemented SNAPLE on top of GraphLab version 2.2 using compiled using GCC 4.8 with C++11 compatibility enabled. We use the Warp engine offered by GraphLab with its default configuration. We run our experiments in a private cluster using two kind of computing nodes: type-I and type-II machines. The type-I nodes are equipped with 2 Intel Xeon L5420 (2.5 Ghz) processors, 4 cores per processor (8 cores per node), 32 GBytes of memory, and Gigabit Ethernet. type-II nodes are more powerful machines designed for big-data workloads with 2 Intel Xeon E5-2660v2 (2.2 Ghz) processors per node, possessing each 10 cores (20 cores per node), 128 GBytes of memory and 10-Gigabit ethernet connection. We deploy our experiments on up to 32 type-I nodes (256 cores) and up to 8 type-II nodes (160 cores). The experiments are executed in type-I machines if not indicated otherwise. All the nodes have access to a shared network file system (NFS) where we store the graphs to be loaded.

Evaluation protocol

For our experiments, we use the set of publicly available datasets described in Table 4. They consist of a set of static directed graphs except gowalla and orkut which are undirected. We transform them into directed by duplicating edges on both directions. This set offering a representative set of comparison points, which includes graphs from different domains, and covers a broad range of sizes, from small instances with less than 1 million edges (gowalla) to very large graphs with over 1 billion edges (twitterrv). [START_REF] Cho | Friendship and mobility: User movement in location-based social networks[END_REF] 196,591 0.95M social network pokec [START_REF] Takac | Data analysis in public social networks[END_REF] 1.6M 30.6M social network orkut [START_REF] Mislove | Measurement and Analysis of Online Social Networks[END_REF] 3M 223M social network livejournal [START_REF] Backstrom | Group formation in large social networks: Membership, growth, and evolution[END_REF] 4.8M 68.9M co-authorship twitter-rv [START_REF] Kwak | What is twitter, a social network or a news media[END_REF] 41M 1.4B microblogging

In order to emulate the prediction process we follow a similar approach to the one taken in in [START_REF] Sarkar | Fast nearest-neighbor search in disk-resident graphs[END_REF]. We randomly remove one outgoing edge from each vertex with |Γ(u)| > 3. After the execution, we obtain Inria k (with k = 5 fixed) predictions for each vertex. Experiments with SNAPLE are parametrized by a score combination (taken from Table 3), a truncation threshold thr Γ , and a sampling parameter k local . Unless indicated otherwise thr Γ = 200. We use Jaccard's coefficient as raw similarity (function f in equation ( 6)). The linear combinator is configured with α = 0.9, which was found to return the best predictions. We use two metrics to compare approaches: recall and execution time. Recall is the proportion of removed edges that are successfully returned by the algorithm. Execution time is measured from when the graph has been successfully loaded (as reported by GraphLab) until after all predictions have been computed. We do not take into consideration the time spent to load the graph as it completely depends on the GraphLab implementation and the network file system, being both aspects outside the scope of this work.

Let us note that recommendation systems are usually evaluated using a second metric, precision, which is the proportion of correct recommendations within the returned answers. Because we have fixed the number of edges removed from each vertex (one), and the number of predictions (k = 5) we return, precision becomes proportional to recall, and is therefore not relevant in our set-up.

Comparison with a direct implementation

We first compare SNAPLE against a direct implementation of Algorithm 1 on GraphLab using Jaccard's coefficient. For a fair comparison, we limit the search of candidates to 2-hop neighborhoods, as in SNAPLE. We call this implementation BASELINE. As discussed in Section 2.4, BASELINE needs to directly compute similarity between every pair of vertices that lie 2 hops away of each other. Because the GAS model only provides access to direct neighbors, BASELINE must propagate and store neighborhood information along every 2-hop paths, resulting in substantial overheads.

We compare BASELINE against 12 configurations of SNAPLE, by using three scoring configurations from Table 3 (linearSum, counter, and PPR), and by varying the truncation threshold thr Γ and the sampling parameter k local between ∞ (in effect no truncation, resp. no sampling) and 20 (a low value to exacerbate the effect of each parameter). We execute BASELINE and the 12 SNAPLE configurations on 4 type-II nodes (80 cores), and apply them to the datasets gowalla, pokec and livejournal . (orkut and twitter-rv cause BASELINE to fail by exhausting the available memory.) Table 5 reports the recall values and execution times we obtain, with the recall gains (resp. speedups) shown in brackets for SNAPLE configurations, computed against BASELINE. The table shows that even without truncation or sampling (thr Γ = k local = ∞), SNAPLE clearly outperforms BASELINE both in terms of recall (which more than doubles on all datasets) and execution times, with speedups ranging from 1.6 to 4.6. Truncation (thr Γ = 20) brings a small improvement in speed-up, while causing recall to go down slightly, be it with or without k local . The sampling parameter k local has the largest impact on the execution time by far, yielding speedups ranging from 15.8 (pokec) up to 109 on gowalla when applied alone (thr Γ = ∞), while having a minimal impact on recall. Adding truncation to k local further improves execution times for the pokec (speed-ups ranging from 19 to 22.7) and livejournal (35.2 to 39.3) (last 3 lines of the table ).

RR n°454

In terms of score configuration, linearSum, counter, and PPR tend to produce similar recall values, with a slight advantage for linearSum (best recalls are shown in bold). The differences in execution times between the three score configuration are within the experimental noise, and not significant. This first experiment demonstrates the flexibility of SNAPLE, which offers a large number of configurations. These results also illustrate the benefits of SNAPLE in terms of recall and execution time, and highlight the importance of sampling to get important speedup while maintaining a good recall. Although the effect of thr Γ is less pronounced than that of k local , truncation remains an important mechanism to deal with highly connected vertices. In Sections 5.5 and 5.7 we return to the impact of both parameters, with a more fine-grained sensitivity analysis of their effects. We otherwise fix thr Γ = 200 for our other experiments.

Scalability

We assess the scalability of SNAPLE by applying the linearSum scoring to livejournal, orkut and twitterrv on varying numbers of type-I and type-II nodes. Figure 5 shows the results we obtain for two values of k local (40 and 80). Other scoring configurations return near-identical results. thr Γ % recall improvement q livejournal orkut twitter-rv (d) Relative recall improvement Figure 6: Relative recall improvement for different thr Γ values when compared with the one obtained using thr Γ = 10 with linear combinator. We highlight the used thr Γ values in the CDF of the output degrees of the analyzed datasets. When thr Γ reaches around 80% of the observations in the CDF there is no longer improvement.

Impact of the truncation threshold thr Γ

The truncation of neighborhoods using thr Γ (Sec. 4) serves two purposes: to limit memory overheads, in particular in the case of densely connected components, and to improve execution times (particularly in conjunction with k local ).

By sampling truncating neighborhoods, we might however loose relevant information, disturb the computation of raw similarities, and finally negatively impact the quality of link predictions. This effect is however limited to vertices whose neighborhood is larger than thr Γ . To shed more light on this phenomenon, Figures 6a, 6b and6c shows the CDF of vertex degrees in orkut, livejournal and twitter-rv, and superimpose five different values of thr Γ (10, 20, 40, 80 and 100, shown as vertical lines). These figures show that already with thr Γ = 100, only a minority of vertices will be truncated across all three datasets, including very small minorities (around 1%) in the case of livejournal and twitter-rv.

This analysis is confirmed by Figure 6d, which shows how recall improves using linearSum and k local = 80, while varying thr Γ from 10 to 100 (recall values are normalized by the recall obtained with thr Γ = 10, and the relative improvement is shown). The impact of thr Γ is strongest on orkut, whose degree distribution varies strongly in the interval of values taken by thr Γ . In all three datasets, recall stabilizes when thr Γ reaches 80, which is when thr Γ covers at least 80% of all vertices in all three graphs. This shows that the impact of thr Γ on recall can be minimized by selecting it appropriately, while limiting the memory impact of the most densely connected vertices.

Impact of the vertex selection mechanism

We limit the number of paths to be explored by sampling the neighborhood of each node to a set with the k local most similar neighbors using the Γ max k local function (Step 2 of Algorithm 2). Discarding vertices reduces the computing time with the inconvenient of reducing the set of explored candidates. This has a direct impact on the recall. Using Γ max k local , we consider the most similar vertices to be good candidates for the sampling. In order to demonstrate that sampling using similarity improves the search of potential vertex candidates, thus recall. We compare our proposed neighbors selection Γ max k local with Γ min k local and Γ rnd k local . As mentioned Γ max k local selects the k local neighbors with the largest similarity, the second those with the smallest similarity and the last one runs a random selection.

Figure 7 shows the recall for various score methods in the livejournal dataset for different k local values. As k local grows the obtained recall converges for the three policies because we expand the search until we explore the same candidates in all cases. However, for small values of k local Γ max k local always gets larger RR n°454 q q q q q q q q q q q q q q q counter linearSum PPR 0.10 k local . This result indicates that using the similarity as a criterion to limit the amount of vertices to explore is particularly effective for small values of k local . Reducing the number of vertices to explore reduces both the computation time and the amount of memory to use.

Impact of the sampling parameter k local

The definition of k local puts an upper bound limit of k 2 local candidate vertices to be scored. Reducing the search space reduces time and storage paying a penalty in recall. However, different scores may get different recall for the same k local value. We have already showed that selecting the most similar vertices Inria we can increase recall when using small k local values. Now we analyze the impact of varying k local in the computing time and recall for the scoring methods proposed in Table 3.

Figure 8 shows the recall and computing time for livejournal and twitter-rv datasets. We reduce the presented figures to these datasets due to the lack of space. The Sum aggregator is exhaustive in the sense that it accumulates all pair-wise similarities, thus taking into account the popularity of a vertex z (i.e. the number of paths connecting u to z) in the final score it returns. This behavior explains why the recall increases together with k local . By contrast, Mean and Geom work differently. Mean improves the recall obtained with Sum for small values of k local (see linearMean). When k local increases the recall obtained with Mean goes down. We believe this may occur because Mean averages an increasing number of pathsimilarities for larger k local , many of which might have low scores. Additionally, averaging eliminates information about the popularity of a vertex, contrarily to Sum. The Geom aggregator show the same pattern in a stronger form, which is probably due to its sensitivity to non-similar vertices (sim (u, z) = 0).

Thoroughly understanding the properties of each scoring configuration would require a longer and deeper analysis outside the scope of this work. On the basis of our current results, we can however propose some guidelines. For scenarios requiring the best predictions, but not necessarily the shortest times, the linearSum score seems to be the best solution. By increasing k local it produces the best recall values, at the cost of higher computation times. For scenarios demanding the best results under a tight time budget, the Mean aggregator with small values of k local appears competitive, and in some cases a better solution that linearSum.

Sensitivity of recall to k and to removed edges

For completeness, we analyze how recall evolve when we vary the number of answers SNAPLE returns. This is shown in Figure 9 for livejournal and pokec, and five score configurations, when k takes the values 5, 10, 15, 20. In this range, the recall increases substantially with k. (The other scores based on the Mean and Geom aggregators follow a similar pattern.)

Similarly, we investigate the sensitivity of recall when we increase the number of edges that is removed per vertex. Removing more edges, we remove paths between vertices, which makes it more difficult for SNAPLE to find relevant vertices. As a result recall decreases (Figure 10). (If a vertex has less edges than the number to be removed, we removed all the edges except one.) The recall decreases proportionally to the number of removed edges. We observe similar patterns when using Mean and Geom aggregators. Removed edges vertex Recall q counter euclSum geomSum linearSum PPR (a) livejournal q q q q q 0.08 0.10 0.12 0.14

Comparison to Cassovary

For exhaustiveness we conclude our evaluation by comparing SNAPLE with a single-machine solution. This comparison serves two objectives: assess the impact of SNAPLE when networking costs disappear, and provide a reference point to gauge the benefits of distribution when processing very large graphs. We use Cassovary [START_REF]Cassovary[END_REF] for this comparison, a multithread in-memory Graph library developed by Twitter.

Cassovary is able to load relative large graphs and can traverse a graph fully allocated into main memory. It has been shown to be a efficient solution for computing random walks [START_REF] Kyrola | Drunkardmob: Billions of random walks on just a pc[END_REF][START_REF] Lofgren | Fast-ppr: Scaling personalized pagerank estimation for large graphs[END_REF]. It is also used in production by Twitter [START_REF] Gupta | Wtf: the who to follow service at twitter[END_REF].

In a first attempt, we implemented the solution described by Algorithm 1 (with the 2-hop optimization). However, neither the recall nor computing time were competitive. We therefore moved on to a multithreaded version of the personalized page rank (PPR) [START_REF] Page | The pagerank citation ranking: Bringing order to the web[END_REF] approximation based on random walks [START_REF] Sarlós | To randomize or not to randomize: Space optimal summaries for hyperlink analysis[END_REF] to improve on these results. For each vertex we run w random walks of depth d. d = 2 reaches the neighbors of a vertex ; d = 3 its neighbors of neighbors and so on. Once the random walks terminates, the k most visited vertices not included into Γ(v) are returned as predictions. Increasing w and d, we force the algorithm to explore a larger number of vertices in a similar way we do when varying k local . We modify w and d configurations in order to find the largest recall in the shortest time. Figure 11 compares the recall and computing time for Cassovary running on a type-II machine when varying w and d on the livejournal and twitter-rv datasets. We observe that increasing d does not necessarily improve recall, with d = 3 yielding recall values very close to that of larger values of d. By contrast, larger values of w tend to yield better recall value, but they also significantly increase the computing time. In the case of twitter-rv, we run an extra configuration with d = 3 and w = 10000 getting 0.06 recall in 90 minutes. Unfortunately, we had to stop other configurations with larger d values and w = 10000 as they took too long to complete. SNAPLE is designed to run on a distributed environment taking advantage of the speed of multiple simultaneous computing nodes. A comparison between SNAPLE on multiple machines and Cassovary on a single machine would therefore not be fair to Cassovary. For that reason, we compare the best Inria results obtained in our previous analysis of Cassovary (best recall in the shortest time) with the results obtained running SNAPLE on a single machine. We use k local = 20, which produces recall values close to that of Cassovary (slightly higher in fact). The results we obtain (Table 6) show that SNAPLE is faster than Cassovary (with speedups of 2.03 and 9.02) while increasing recall. This demonstrates that even on single machine deployments SNAPLE provides a competitive solution in terms of both prediction quality and execution time.

We can also use the results obtained with Cassovary to assess the benefits of distribution when processing very large graphs: the recall obtained by Cassovary on twitter-rv (0.06, Table 6) is obtained by SNAPLE in 177s (2min57s) when using linearSum with k local = 5 on 256 type-I cores (32 type-I machines), as reported by Figure 8a (right-hand chart). This corresponds to a speedup of 30.62 against Cassovary, while only using 12.8 more cores.

Related work

Processing large graphs in a single machine has been done with success in [START_REF] Gupta | Wtf: the who to follow service at twitter[END_REF]. Other recent approaches such as GraphChi [START_REF] Kyrola | Graphchi: Large-scale graph computation on just a pc[END_REF] or X-Stream [START_REF] Roy | X-stream: Edge-centric graph processing using streaming partitions[END_REF] propose the utilization of specific data structures and secondary storage to make feasible processing graphs not fitting into memory. These approaches have been shown to be competitive and resource efficient solutions. However, they are not as scalable as distributed solutions neither take advantage of the vast amount of resources offered by the cloud.

Besides GAS-based engines discussed in Section 2, there is a large number of works aiming at providing abstraction models that easily permit to work with large amounts of data in a distributed manner. The BSP (Bulk Synchronous Parallel) [START_REF] Valiant | A bridging model for parallel computation[END_REF] communication model runs multiple threads dispatching messages to other vertices. The computation is done in supersteps with synchronization barriers between each. The complexity of programming communication models like BSP have driven researchers to propose distributed shared memory solutions [START_REF] Bershad | The midway distributed shared memory system[END_REF][START_REF] Johnson | Crl: High-performance all-software distributed shared memory[END_REF]. However, the cost of remote memory access make these solutions unfeasible for high-performance. An improvement has been proposed through PGAS (Parallel Global Address Space) [START_REF] Basumallik | Programming distributed memory sytems using openmp[END_REF] by reducing the number of remote memory accesses. Recently, the popularity of MapReduce solutions has inspired high-level languages such as Hive [START_REF] Thusoo | Hive: A warehousing solution over a map-reduce framework[END_REF] or Pig [START_REF] Olston | Pig latin: A not-so-foreign language for data processing[END_REF]. However, none of these solutions consider the lack of locality when traversing the graphs [START_REF] Lumsdaine | Challenges in parallel graph processing[END_REF].

The current ecosystem of graph engines is persistently growing. Projects such as GraphX [START_REF] Gonzalez | Graphx: Graph processing in a distributed dataflow framework[END_REF] offer abstraction layers that permit to combine different graph engines in graph workload analysis abstracting the developer from the details of each engine.

Conclusion

In this paper we have presented the design, implementation and evaluation of SNAPLE, a highlyscalable approach to the link-prediction problem optimized for the gather-apply-scatter (GAS) model of distributed graph engines. We have provided an exhaustive evaluation of our prototype in a cluster using a representative array of large publicly available datasets. SNAPLE is able to compute the predictions of a graph containing 1.4 billion edges in less than 3 minutes when other naive GraphLab solutions fail due to resource exhaustion. Additionally, we demonstrate that SNAPLE has an over-linear speedup of 30 when compare with a state-of-the-art non-distributed solution while improving prediction quality.

Our work opens several exciting research paths we plan to explore. One such path involved the extension of SNAPLE to supervised link-prediction strategies, which may improve recall while taking advantage of distributed computing. We also would like to port SNAPLE to other distributed graph processing platforms such as Giraph [START_REF]Apache giraph[END_REF], Bagel [START_REF] Bagel | [END_REF] or Stinger [START_REF] Ediger | Stinger: High performance data structure for streaming graphs[END_REF] to provide more comparison points between these platforms. Finally, we plan to perform a more in-depth analysis of the data structures used in our prototype to understand how reduce operations can be further improved, and help reduce latency.

Algorithm 1 2 :

 12 Unsupervised top k link-predictionRequire: k,score1: for u ∈ V do prediction u ← argtop k z∈V \Γ G (u)

Figure 3 :

 3 Figure 3: Example of scores obtained using different aggregators and the linear combinator with α = 0.5. Edge weights indicate the similarity between vertices.

Figure 4 :

 4 Figure 4: Illustrating Algorithm 2. In Step 1, each vertex constructs its list of neighbors Γ. This list of neighbors is propagated to adjacent vertices (Step 2) to compute raw similarities sims[•] (for a: with c, d, and b). Only the k local most similar vertices are kept. Finally (Step 3), raw similarities are used to compute path similarities, which are then combined into final scores.

12: gather 3 ( 13 :; 14 :z

 31314 D u , D (u,v) , D v ): if v ∈ D u .sims.keys then return / 0 Γu ← D u . Γu ; Γ max v ← D v .sims.keys 15: return (z, D u .sims[v] ⊗ D v .sims[z], 1) z ∈ Γ max v \ Γu 16: sum 3 (γ 1 ,γ 2 ): return merge(⊕ pre , γ 1 , γ 2 ) Comp.Σ 3 17: apply 3 (D u , Σ 3 ): , n z ) ∈ Σ 3 do score[z] ← ⊕ post (s pre z , n z ) 20: D u .predicted ← argtop k z∈score.keys score[z]

  k local = 80 and type-II

Figure 5 :

 5 Figure 5: SNAPLE scales linearly graph sizes (measured in edges). Execution times are reported for both type-I and type-II nodes, and for k local ∈ {40, 80}. Missing points indicate configurations not fitting into memory.

  SNAPLE scales linearly with graph sizes (measured in edges), for both values of k local , and on both type-I and type-II nodes. Doubling k local increases the number of paths to be considered, and increases the execution time by 70%. The required memory also increases, which makes it impossible to run the experiment on twitter-rv with k local = 80 and only 8 type-I machines (64 cores). The most exhaustive experiment we run processes the twitter-rv dataset (1.4 billion edges) with k local = 80 in less than 10 minutes (585s) using 256 type-I cores (32 machines) or in a similar time with 160 type-II cores (8 machines). These executions yield a recall of 0.093. (A value we return to when we analyze the effect of k local on recall and execution time in Sec. 5.7.) Inria

Figure 7 :Figure 8 :

 78 Figure 7: Recall obtained for three different neighbor sampling policies using k local = 5, 10, 20, 40, 80 with the livejournal dataset. Selecting the most similar vertices (Γ max k local ) improves the recall, in particular for small values of k local .

Figure 9 :

 9 Figure 9: Evolution of recall when increasing the number of recommended links k with k local = 80.

Figure 10 :

 10 Figure 10: Evolution of recall when increasing the number of removed edges per vertex with k local = 80.

Figure 11 :

 11 Figure 11: Recall and computing time using a stand-alone link-prediction solution on top of Cassovary using random walks to emulate PPR. Both solutions are run on a type-II machine with w = 10, 100, 1000.

Table 3 :

 3 Examples of score(u, z) combinations in SNAPLE's scoring framework sim(u, v) combinator (⊗) aggregator (⊕) score name

	Jaccard	linear		linearSum
	Jaccard	eucl		euclSum
	Jaccard	geom	Sum	geomSum
	1/|Γ v |	sum		PPR
	-	count		counter
	Jaccard	linear		linearMean
	Jaccard	eucl	Mean	euclMean
	Jaccard	geom		geomMean
	Jaccard	linear		linearGeom
	Jaccard	eucl	Geom	euclGeom
	Jaccard	geom		geomGeom

  2 (u), but only those paths going through the k local edges with the highest similarity values at individual vertices. Said differently, we sample down

						e
			0		0 . 1
		h		b	0.3	f
	0	a	0 . 2 0.1 0 . 5	c	0 . 2 0.2 0 . 2 0 . 3	g
				d
		score(a, e) score(a, f ) score(a, g)
	linearSum		0.3		0.6	0.75
	linearMean		0.15		0.3	0.25
	linearGeom		0		0.28	0.24

Table 4 :

 4 The datasets used in the evaluation

	dataset	|V |	|E|	domain
	gowalla

Table 5 :

 5 SNAPLE clearly outperforms a direct implementation of similarity-based link-prediction on GraphLab, both in terms of recall and execution time. The results were obtained on 4 type-II nodes (80 cores). Gains (for recall), and speedups (for time) are shown in brackets.

					dataset
		score(u, z)	gowalla	pokec	livejournal
			recall	time (s)	recall	time (s)	recall	time (s)
		BASELINE	0.12	119.9	0.05	213.8	0.12	1010.7
					thr Γ = ∞, k local = ∞
		linearSum	0.28 (2.3)	72.8 (1.6)	0.14 (2.8)	60.2 (3.5)	0.31 (2.5) 224.5 (4.5)
		counter	0.33 (2.7)	63.1 (1.9)	0.12 (2.4)	59.7 (3.5)	0.28 (2.3) 218.0 (4.6)
		PPR	0.26 (2.1)	68.0 (1.7)	0.12 (2.4)	56.6 (3.7)	0.30 (2.5) 222.5 (4.5)
					thr Γ = 20, k local = ∞
		linearSum	0.26 (2.1)	66.1 (1.8)	0.12 (2.4)	58.3 (3.6)	0.27 (2.2) 213.8 (4.7)
	SNAPLE	counter PPR linearSum	0.23 (1.9) 0.24 (2.0) 0.28 (2.3)	69.8 (1.7) 68.5 (1.7) 1.1 (109)	0.11 (2.2) 0.11 (2.2) thr Γ = ∞, k local = 20 60.0 (3.5) 60.1 (3.5) 0.13 (2.6) 12.8 (16.7) 0.30 (2.5) 32.5 (31.0) 0.26 (2.1) 211.8 (4.7) 0.28 (2.3) 211.5 (4.7)
		counter	0.24 (2.1)	1.0 (119)	0.12 (2.4) 13.0 (16.4) 0.27 (2.2) 29.6 (34.1)
		PPR	0.26 (2.1)	1.1 (109)	0.11 (2.2) 13.5 (15.8) 0.29 (2.4) 38.3 (26.3)
					thr Γ = 20, k local = 20
		linearSum	0.26 (2.1) 1.1 (109.0) 0.11 (2.2)	9.4 (22.7)	0.26 (2.1) 28.7 (35.2)
		counter	0.22 (1.8) 1.1 (109.0) 0.10 (2.0) 11.0 (19.4) 0.24 (2.0) 26.4 (38.2)
		PPR	0.25 (2.0) 1.2 ( 99.9) 0.10 (2.0) 11.2 (19.0) 0.26 (2.1) 25.7 (39.3)

Table 6 :

 6 SNAPLE also outperforms a state-of-the-art single-machine solution (results obtained on one type-II node)

	dataset	CASSOVARY recall time(s) recall time(s) speedup SNAPLE
	livejournal	0.24	93	0.30	45.8	2.03
	twitter-rv	0.06	5420	0.08	600.7	9.02

RR n°454

Inria

We limit ourselves to 2-hop paths, but this approach can be extended to longer paths by recursively applying ⊗ to the raw similarities of individual edges (in functional terms, essentially executing a fold operation on the raw similarity values along the path connecting u to z).RR n°454