Compartilhar via

Guia do usuário do GraphFrames – Scala

Este artigo demonstra exemplos do Guia do Usuário do GraphFrames.

import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.graphframes._

Criando GraphFrames

Você pode criar GraphFrames a partir de DataFrames de vértices e arestas.

  • DataFrame do Vértice: um DataFrame de vértice deve conter uma coluna especial nomeada id que especifica IDs exclusivas para cada vértice no grafo.
  • DataFrame de Borda: um DataFrame de borda deve conter duas colunas especiais: src (ID de vértice de origem da borda) e dst (ID de vértice de destino da borda).

Ambos os DataFrames podem ter outras colunas arbitrárias. Essas colunas podem representar atributos de vértice e borda.

Criar os vértices e as bordas

// Vertex DataFrame
val v = spark.createDataFrame(List(
  ("a", "Alice", 34),
  ("b", "Bob", 36),
  ("c", "Charlie", 30),
  ("d", "David", 29),
  ("e", "Esther", 32),
  ("f", "Fanny", 36),
  ("g", "Gabby", 60)
)).toDF("id", "name", "age")
// Edge DataFrame
val e = spark.createDataFrame(List(
  ("a", "b", "friend"),
  ("b", "c", "follow"),
  ("c", "b", "follow"),
  ("f", "c", "follow"),
  ("e", "f", "follow"),
  ("e", "d", "friend"),
  ("d", "a", "friend"),
  ("a", "e", "friend")
)).toDF("src", "dst", "relationship")

Vamos criar um grafo com base nesses vértices e nestas bordas:

val g = GraphFrame(v, e)

// This example graph also comes with the GraphFrames package.
// val g = examples.Graphs.friends

Consultas básicas de grafos e DataFrames

GraphFrames fornecem consultas simples sobre grafos, como o grau dos nós.

Além disso, como GraphFrames representa grafos como pares de DataFrames de vértices e arestas, é fácil fazer consultas avançadas diretamente nos DataFrames de vértices e arestas. Esses DataFrames estão disponíveis como vértices e campos de bordas no GraphFrame.

display(g.vertices)

display(g.edges)

O grau de entrada dos vértices:

display(g.inDegrees)

O grau de saída dos vértices é:

display(g.outDegrees)

O grau dos vértices:

display(g.degrees)

Você pode executar consultas diretamente no DataFrame de vértices. Por exemplo, podemos encontrar a idade da pessoa mais jovem no grafo:

val youngest = g.vertices.groupBy().min("age")
display(youngest)

Da mesma forma, você pode executar consultas no DataFrame de bordas. Por exemplo, vamos contar o número de relações 'follow' no grafo:

val numFollows = g.edges.filter("relationship = 'follow'").count()

Identificação de motivos

Crie relações mais complexas envolvendo bordas e vértices usando motivos. A célula a seguir localiza os pares de vértices com bordas em ambas as direções entre eles. O resultado é um DataFrame, no qual os nomes de coluna são chaves de motivo.

Confira o Guia do Usuário do GraphFrame para obter mais detalhes sobre a API.

// Search for pairs of vertices with edges in both directions between them.
val motifs = g.find("(a)-[e]->(b); (b)-[e2]->(a)")
display(motifs)

Como o resultado é um DataFrame, você pode criar consultas mais complexas com base no padrão. Vamos encontrar todas as relações recíprocas em que uma pessoa tem mais de 30 anos:

val filtered = motifs.filter("b.age > 30")
display(filtered)

Consultas com estado

A maioria das consultas de "motif" não possuem estado e são simples de expressar, como nos exemplos acima. Os exemplos a seguir demonstram consultas mais complexas que carregam o estado ao longo de um caminho no motivo. Expresse essas consultas combinando a busca de padrões do GraphFrame com filtros no resultado, onde os filtros utilizam operações de sequência para construir uma série de colunas do DataFrame.

Por exemplo, suponha que você queira identificar uma cadeia de 4 vértices com alguma propriedade definida por uma sequência de funções. Ou seja, entre cadeias de 4 vértices a->b->c->d, identifique o subconjunto de cadeias que correspondem a este filtro complexo:

  • Inicializar estado no caminho.
  • Atualizar o estado com base no vértice a.
  • Atualizar o estado com base no vértice b.
  • Etc. em relação a c e d.
  • Se o estado final corresponder a alguma condição, o filtro aceitará a cadeia.

Os snippets de código a seguir demonstram esse processo, em que identificamos cadeias de 4 vértices, de modo que pelo menos 2 das 3 bordas são relações "amigas". Neste exemplo, o estado é a contagem atual de bordas "amigas"; em geral, pode ser qualquer coluna DataFrame.

// Find chains of 4 vertices.
val chain4 = g.find("(a)-[ab]->(b); (b)-[bc]->(c); (c)-[cd]->(d)")

// Query on sequence, with state (cnt)
//  (a) Define method for updating state given the next element of the motif.
def sumFriends(cnt: Column, relationship: Column): Column = {
  when(relationship === "friend", cnt + 1).otherwise(cnt)
}
//  (b) Use sequence operation to apply method to sequence of elements in motif.
//      In this case, the elements are the 3 edges.
val condition = Seq("ab", "bc", "cd").
  foldLeft(lit(0))((cnt, e) => sumFriends(cnt, col(e)("relationship")))
//  (c) Apply filter to DataFrame.
val chainWith2Friends2 = chain4.where(condition >= 2)
display(chainWith2Friends2)

Subgrafos

O GraphFrames fornece APIs para criar subgrafos filtrando bordas e vértices. Esses filtros podem ser compostos juntos. Por exemplo, o subgrafo a seguir contém apenas pessoas que são amigas e que têm mais de 30 anos.

// Select subgraph of users older than 30, and edges of type "friend"
val g2 = g
  .filterEdges("relationship = 'friend'")
  .filterVertices("age > 30")
  .dropIsolatedVertices()

Filtros trigêmeos complexos

O exemplo a seguir mostra como selecionar um subgrafo com base em filtros trigêmeos que operam em uma borda e seus vértices "src" e "dst". Estender este exemplo para ir além dos trigêmeos usando motivos mais complexos é simples.

// Select subgraph based on edges "e" of type "follow"
// pointing from a younger user "a" to an older user "b".
val paths = g.find("(a)-[e]->(b)")
  .filter("e.relationship = 'follow'")
  .filter("a.age < b.age")
// "paths" contains vertex info. Extract the edges.
val e2 = paths.select("e.src", "e.dst", "e.relationship")
// In Spark 1.5+, the user may simplify this call:
//  val e2 = paths.select("e.*")

// Construct the subgraph
val g2 = GraphFrame(g.vertices, e2)

display(g2.vertices)

display(g2.edges)

Algoritmos de grafo padrão

Esta seção descreve os algoritmos de grafo padrão integrados ao GraphFrames.

Pesquisa de amplitude (BFS)

Pesquise por "Esther" para usuários de 32 anos < .

val paths: DataFrame = g.bfs.fromExpr("name = 'Esther'").toExpr("age < 32").run()
display(paths)

A pesquisa também pode limitar os filtros de aresta e os comprimentos máximos dos caminhos.

val filteredPaths = g.bfs.fromExpr("name = 'Esther'").toExpr("age < 32")
  .edgeFilter("relationship != 'friend'")
  .maxPathLength(3)
  .run()
display(filteredPaths)

Componentes conectados

Calcule o pertencimento a componentes conectados de cada vértice e retorne um grafo com cada vértice atribuído a um ID de componente.

val result = g.connectedComponents.run() // doesn't work on Spark 1.4
display(result)

Componentes fortemente conectados

Compute o componente fortemente conectado (SCC) de cada vértice e retorne um grafo com cada vértice atribuído ao SCC que contém esse vértice.

val result = g.stronglyConnectedComponents.maxIter(10).run()
display(result.orderBy("component"))

Propagação de rótulo

Execute o Algoritmo de Propagação de Rótulo estático para detectar comunidades em redes.

Cada nó na rede é inicialmente atribuído à sua própria comunidade. A cada superstep, os nós enviam sua afiliação comunitária a todos os vizinhos e atualizam seu estado para o modo de afiliação da comunidade de mensagens de entrada.

LPA é um algoritmo de detecção de comunidade padrão para grafos. É barato computacionalmente, embora (1) a convergência não seja garantida e (2) alguém possa acabar com soluções triviais (todos os nós se identificando em uma única comunidade).

val result = g.labelPropagation.maxIter(5).run()
display(result.orderBy("label"))

PageRank

Identifique vértices importantes em um grafo com base em conexões.

// Run PageRank until convergence to tolerance "tol".
val results = g.pageRank.resetProbability(0.15).tol(0.01).run()
display(results.vertices)

display(results.edges)

// Run PageRank for a fixed number of iterations.
val results2 = g.pageRank.resetProbability(0.15).maxIter(10).run()
display(results2.vertices)

// Run PageRank personalized for vertex "a"
val results3 = g.pageRank.resetProbability(0.15).maxIter(10).sourceId("a").run()
display(results3.vertices)

Caminhos mais curtos

Calcula os caminhos mais curtos para o dado conjunto de vértices de referência, em que os pontos de referência são especificados por IDs de vértice.

val paths = g.shortestPaths.landmarks(Seq("a", "d")).run()
display(paths)

Contagem de triângulos

Calcula o número de triângulos que passam por cada vértice.

import org.graphframes.examples
val g: GraphFrame = examples.Graphs.friends  // get example graph

val results = g.triangleCount.run()
results.select("id", "count").show()
  • Last updated on