L'algorithme PageRank évalue l'importance relative des nœuds dans un graphe orienté, en se basant sur la structure des liens entrants. Initialement conçu pour classser les pages web, il est aujourd’hui largement utilisé dans l’analyse de réseaux sociaux, de citations ou de tout graphe où l’influence d’un nœud dépend de ses connexions.

Dans Spark GraphX, la méthode pageRank(tolérance) permet de calculer itérativement le score PageRank de chaque sommet jusqu’à ce que la variation entre deux itérations soit inférieure à la tolérance spécifiée. Le résultat est un graphe dont les attirbuts des sommets sont remplacés par leur score PageRank.

Analyse de l'influence dans un réseau social

Le code suivent construit un graphe à partir de données explicites de sommets (utilisateurs avec âge) et d’arêtes (relations de suivi), puis calcule le PageRank pour identifier les utilisateurs les plus influents :

import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object InfluenceAnalysis {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("PageRankExample")
      .master("local[*]")
      .getOrCreate()

    val users = Array(
      (1L, ("Alice", 28)),
      (2L, ("Bob", 27)),
      (3L, ("Charlie", 65)),
      (4L, ("David", 42)),
      (5L, ("Ed", 55)),
      (6L, ("Fran", 50))
    )
    val vertices: RDD[(Long, (String, Int))] = spark.sparkContext.parallelize(users)

    val follows = Array(
      Edge(2L, 1L, 7),
      Edge(2L, 4L, 2),
      Edge(3L, 2L, 4),
      Edge(3L, 6L, 3),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 2),
      Edge(5L, 3L, 8),
      Edge(5L, 6L, 3)
    )
    val edges = spark.sparkContext.parallelize(follows)

    val socialGraph = Graph(vertices, edges)

    val rankedUsers = socialGraph.pageRank(0.0001).vertices
    rankedUsers.sortBy(_._2, ascending = false).collect().foreach(println)

    println("=== Degrés entrants ===")
    socialGraph.inDegrees.collect().foreach(println)
    println("=== Degrés sortants ===")
    socialGraph.outDegrees.collect().foreach(println)
  }
}

Évaluation de l'importance à partir de fichiers externes

Dans ce scénario, les arêtes sont stockées dans un fichier texte au format source cible, et les identifiants des sommets sont associés à des noms via un second fichier. On utilise GraphLoader.edgeListFile pour charger le graphe sans attributs initiaux, puis on joint les résultats du PageRank aux noms des utilisateurs :

Fichier users.csv :

1,Mike
2,Nancy
3,Selina
4,Tom

Fichier edges.txt :

1 2
2 2
3 2
4 2
3 1
2 2
3 1
2 2
1 4

import org.apache.spark.graphx.GraphLoader
import org.apache.spark.sql.SparkSession

object ExternalPageRank {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("ExternalGraphPageRank")
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext

    val rawGraph = GraphLoader.edgeListFile(sc, "src/main/resources/graphx01/edges.txt").cache()
    val pageRanks = rawGraph.pageRank(0.001).vertices

    val nameMap = sc.textFile("src/main/resources/graphx01/users.csv")
      .map(line => {
        val parts = line.split(",")
        (parts(0).toLong, parts(1))
      })

    pageRanks.join(nameMap)
      .map { case (id, (score, name)) => (name, score) }
      .collect()
      .foreach(println)
  }
}