GraphFrames guia do usuário Scala
Este artigo demonstra exemplos do guia do usuárioGraphFrames.
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.graphframes._
Criando GraphFrames
O senhor pode criar GraphFrames a partir de DataFrames de vértices e bordas.
- DataFrame de vértice: Um DataFrame de vértice deve conter uma coluna especial chamada
idque especifica IDs exclusivos para cada vértice no gráfico. - DataFrame de borda: Um DataFrame de borda deve conter duas colunas especiais:
src(ID do vértice de origem da borda) edst(ID do vértice de destino da borda ).
Ambos os DataFrames podem ter outras colunas arbitrárias. Essas colunas podem representam atributos de vértice e aresta.
Crie 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 gráfico com esses vértices e essas bordas:
val g = GraphFrame(v, e)
// This example graph also comes with the GraphFrames package.
// val g = examples.Graphs.friends
Gráfico básico e consultas em DataFrame
GraphFrames fornecer consultas gráficas simples, como o grau do nó.
Além disso, como o site GraphFrames representa o gráfico como pares de vértices e arestas DataFrames, é fácil fazer consultas avançadas diretamente no vértice e na aresta DataFrames. 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)
O senhor pode executar consultas diretamente nos vértices DataFrame. Por exemplo, nós podemos encontrar a idade da pessoa mais jovem no gráfico:
val youngest = g.vertices.groupBy().min("age")
display(youngest)
Da mesma forma, o senhor pode executar consultas nas bordas DataFrame. Por exemplo, vamos contar o número de relações "follow" no gráfico:
val numFollows = g.edges.filter("relationship = 'follow'").count()
Descoberta de motivos
Crie relações mais complexas envolvendo arestas e vértices usando motivos. A célula a seguir encontra os pares de vértices com bordas em ambas as direções entre eles. O resultado é um DataFrame, no qual os nomes das colunas são chaves de motivo.
Confira o guia do usuário do GraphFrames 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, o senhor pode criar consultas mais complexas em cima do motivo. Vamos encontrar todas as relações recíprocas nas quais uma pessoa tem mais de 30 anos:
val filtered = motifs.filter("b.age > 30")
display(filtered)
Consultas com estado
A maioria query de motivos não tem estado e é simples de expressar, como nos exemplos acima. Os próximos exemplos demonstram query mais complexas que carregam o estado ao longo de um caminho no motivo. Expresse essas query combinando a localização do motivo GraphFrames com filtros no resultado, onde os filtros usam operações de sequência para construir uma série de colunas 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 correspondentes
esse filtro complexo:
- Inicialize o estado no caminho.
- Atualize o estado com base no vértice a.
- Atualize o estado com base no vértice b.
- Etc. para c e d.
- Se o estado final corresponder a alguma condição, o filtro aceitará a cadeia.
Os trechos de código a seguir demonstram esse processo, onde identificamos cadeias de 4 vértices de forma que pelo menos 2 das 3 arestas sejam “amigas” relacionamentos. Neste exemplo, o estado é a contagem atual de bordas "amigas"; em geral, poderia ser qualquer coluna do 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)
Subgráficos
O GraphFrames fornece APIs para criar subgráficos filtrando bordas e vértices. Esses filtros podem compor juntos. Por exemplo, o subgráfico a seguir contém somente pessoas que são amigas e que têm mais de 30 anos de idade.
// 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 subgráfico com base em filtros tripletos que operam em uma borda e seu “src”. e vértices “dst”. Estender esse 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 gráfico padrão
Esta seção descreve os algoritmos de gráfico padrão incorporados ao GraphFrames.
Pesquisa ampla (BFS)
Pesquise em “Esther” por usuários de idade\ < 32.
val paths: DataFrame = g.bfs.fromExpr("name = 'Esther'").toExpr("age < 32").run()
display(paths)
A pesquisa também pode limitar os filtros de borda 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
calcula a associação de componentes conectados de cada vértice e retorna um gráfico com um ID de componente atribuído a cada vértice.
val result = g.connectedComponents.run() // doesn't work on Spark 1.4
display(result)
Componentes fortemente conectados
calcula o componente fortemente conectado (SCC) de cada vértice e retorna um gráfico com cada vértice atribuído ao SCC que o contém.
val result = g.stronglyConnectedComponents.maxIter(10).run()
display(result.orderBy("component"))
rótulo propagation
execução do rótulo estático Algoritmo de propagação para detecção de comunidade em redes .
Cada nó da rede é inicialmente atribuído à sua própria comunidade. A cada superpasso, os nós enviam sua afiliação à comunidade para todos os vizinhos e atualizam seu estado para o modo de afiliação à comunidade das mensagens recebidas.
O LPA é um algoritmo de detecção de comunidade padrão para gráficos. É barato do ponto de vista computacional, embora (1) a convergência não seja garantida e (2) seja possível acabar com soluções triviais (todos os nós se identificam em uma única comunidade).
val result = g.labelPropagation.maxIter(5).run()
display(result.orderBy("label"))
PageRank
Identificar vértices importantes em um gráfico 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 conjunto dado de vértices de referência, em que as referências são especificadas pelo ID do vértice.
val paths = g.shortestPaths.landmarks(Seq("a", "d")).run()
display(paths)
Contagem de triângulos
calcular 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()