Este é o sexto artigo da série "Big Data com Apache Spark". Por favor, veja também a Parte 1: Introdução, Parte 2: Spark SQL, Parte 3: Spark Streaming, Parte 4: Spark Machine Learning, Parte 5: Spark ML Data Pipelines.

Big data vem em diferentes formas e tamanhos. Podem ser dados em batch que precisam ser processados ​​offline - processando um grande dataset e gerando os resultados e insights em um momento posterior. Ou os dados podem ser recebidos via stream em tempo real e precisam ser processados ​​assim que possível para gerar insights de dados quase instantaneamente.

Vimos como o Apache Spark pode ser usado para processar dados em batch (Spark Core), bem como dados em tempo real (Spark Streaming).

Às vezes, os dados com os quais precisamos lidar estão conectados por natureza. Por exemplo, em um aplicativo de mídia social, temos entidades como usuários, artigos e curtidas, que precisamos gerenciar e processar como uma única unidade lógica de dados. Este tipo de dado é chamado de grafo e requer técnicas analíticas e abordagens diferentes do processamento de dados tradicional.

Este artigo apresenta como processar grafos usando a biblioteca GraphX ​​do Spark. Primeiro, vamos ver o que é um grafo e por que é essencial processar esse tipo de dado em aplicações corporativas de Big Data.

Grafo

Existem três tópicos diferentes a serem abordados quando discutimos tecnologias relacionadas a grafos:

• Bancos de dados orientados a grafos;
• Análise de dados de grafos;
• Visualização de grafos.

Vamos discutir esses tópicos brevemente para aprender como eles diferem e como se complementam para nos ajudar a desenvolver uma arquitetura abrangente de análise e processamento de Big Data baseada em grafos.

Banco de dados orientados a grafos

Ao contrário dos modelos de dados tradicionais, os modelos de grafos têm como elementos principais as entidades e os relacionamentos entre essas entidades. Ao trabalhar com dados em grafos, estamos interessados nas entidades e nas conexões entre elas.

Por exemplo, se trabalhamos em uma aplicação de rede social, estaremos interessados nos detalhes de um usuário específico (digamos, John), mas também queremos modelar, armazenar e recuperar quaisquer associações entre esse usuário e outros usuários na rede. Exemplos dessas associações são "John é amigo de Mike" ou "John leu o livro escrito por Bob".

É importante lembrar que os dados de grafos do mundo real usados nas aplicações são dinâmicos por natureza e mudam com o tempo.

Os bancos de dados orientados a grafos nos permitem descobrir padrões que geralmente são difíceis de detectar usando modelos de dados e abordagens analíticas tradicionais. Alguns exemplos de bancos de dados orientados a grafos, são: Neo4j, DataStax Enterprise Graph, AllegroGraph, InfiniteGraph e OrientDB.

Modelando dados em grafos

A modelagem de dados em grafos inclui definir os nós (também conhecidos como vértices), os relacionamentos (também conhecidos como arestas) e os rótulos para esses nós e relacionamentos.

Os bancos de dados de grafos são baseados no que Jim Webber, da Neo Technologies, chama de "modelagem orientada a consultas", o que significa que o modelo de dados está aberto a especialistas de domínio, ao invés de apenas especialistas em banco de dados, e suporta a colaboração em equipe para modelagem e evolução.

Bancos de dados baseados em grafos como o Neo4J fornecem uma linguagem de consulta (Cypher no caso do Neo4j) para gerenciar os grafos armazenados no banco de dados.

Processando os dados dos grafos

O processamento de grafos inclui principalmente percorrer o grafo para encontrar nós específicos no dataset do grafo que correspondem aos padrões especificados e, em seguida, localizar os nós associados e os relacionamentos nos dados para que possamos ver os padrões de conexão entre diferentes entidades.

O pipeline de processamento de dados normalmente inclui as seguintes etapas:

• pré-processamento de dados (que inclui carregamento, transformação e filtragem);
• criação do grafo;
• análise;
• pós-processamento.

Uma ferramenta de análise de grafo deve fornecer a flexibilidade de trabalhar com grafos e coleções para que possamos combinar tarefas de análise de dados como ETL, análise exploratória e computação iterativa do grafo em um único sistema, sem ter que usar várias estruturas e ferramentas diferentes.

Existem várias estruturas que podemos usar para processar os grafos e executar análises preditivas sobre os dados. Isso inclui Spark GraphX, Gelly do Apache Flink e GraphLab Create.

Neste artigo, vamos nos concentrar no Spark GraphX ​​para analisar dados de grafos.

Existem também vários geradores de grafos diferentes, conforme observado na documentação do Gelly, como grafo de ciclo, grafo de grade, grafo de hipercubo, grafo de caminho e grafo de estrela.

Visualização de grafos

Assim que começarmos a armazenar dados conectados em um banco de dados orientado a grafos e precisamos analisar os dados dos grafos, precisamos de ferramentas para visualizar os padrões nos relacionamentos entre as entidades de dados. Sem ferramentas de visualização, os esforços de análise de dados não são completos.

As ferramentas de visualização de grafos incluem D3.js, Linkurious e GraphLab Canvas.

Casos de uso de grafo

Há diversos casos de uso para os quais os bancos de dados orientados a grafos são mais adequados do que soluções como bancos de dados relacionais ou outros armazenamentos de dados NoSQL. Alguns desses casos de uso incluem o seguinte.

Recomendações e personalização: a análise de grafos pode ser usada para gerar modelos de recomendação e personalização para clientes, e os insights encontrados na análise dos dados podem influenciar decisões importantes. Isso ajuda as empresas a persuadir os clientes a comprar seus produtos. Essa análise também ajuda na estratégia de marketing e na avaliação do comportamento do atendimento ao cliente.

Detecção de fraude: as soluções de grafos também ajudam a revelar transações fraudulentas em um aplicativo de processamento de pagamentos, com base em dados conectados que incluem entidades como usuários, produtos, transações e eventos. Este post no blog phData descreve uma aplicação de exemplo que usa o Spark GraphX ​​para detecção de fraude em comunicação por telefone usando o algoritmo PageRank nos metadados.

Modelagem de assuntos: inclui técnicas que agrupam documentos e extraem representações dos assuntos desses documentos.

Detecção de comunidade: o site Alibaba.com usa técnicas de análise de grafos, como detecção de comunidade, para resolver problemas de comércio eletrônico.

Desempenho de voo: outros casos de uso, conforme discutido neste post do blog do Databricks, analisam dados de desempenho de voo organizados em estruturas de grafs para revelar estatísticas como: classificação de aeroportos, desempenho pontual e caminhos mais curtos entre cidades.

Distância mais curta: descobrir as distâncias e caminhos mais curtos também é útil em aplicativos de rede social. Isso pode ser usado para medir a relevância de um determinado usuário na rede. Usuários com distâncias curtas menores são mais relevantes do que usuários mais distantes.

Spark GraphX

O GraphX é uma API do Apache Spark para grafos e computação paralela de grafos. Ele estende o Spark RDD, introduzindo uma nova abstração do grafo: um multigrafo direcionado com propriedades anexadas a cada vértice e aresta.

A biblioteca GraphX fornece funções para os grafos como: subgraph, joinVertices e aggregateMessages para transformar os grafos. Ele fornece várias maneiras de construir um grafo a partir de uma coleção de vértices e arestas em um RDD ou no disco. O GraphX também inclui vários algoritmos e construtores de grafos para usar e realizar análises nos grafos. Discutiremos os algoritmos dos grafos posteriormente nesta seção.

A Figura 1 mostra onde o GraphX se encaixa no ecossistema Apache Spark.

Figura 1. O ecossistema Spark e a biblioteca GraphX.

O GraphX facilita as análises em grafos com operadores e algoritmos integrados. Também nos permite armazenar e remover o cache dos grafos para evitar recálculo quando precisamos chamar um grafo várias vezes.

Algumas funções de grafos disponíveis no GraphX estão listados na Tabela 1.

Tabela 1. Operadores de grafo do Spark GraphX.

Veremos esses operadores em detalhes na seção "Aplicação de exemplo", quando executarmos algoritmos GraphX em diferentes datasets de rede social.

GraphFrames

O GraphFrames, uma nova ferramenta de processamento de grafos do Spark, integra os recursos como pattern matching e algoritmos de grafo com o Spark SQL. Os vértices e arestas são representados como DataFrames em vez de objetos RDD.

O GraphFrames simplifica o pipeline de análise dos grafos e otimiza as consultas nos grafos e nos dados relacionados. Ele oferece algumas vantagens sobre o processamento de grafos feitos com RDD:

• Suporte a Python e Java, além das APIs Scala, permitindo usar os algoritmos do GraphX em todas as três linguagens;
• Capacidade de usar consultas avançada usando Spark SQL e DataFrames. O plano de consulta dos grafos usa views materializadas para melhorar o desempenho da consulta;
• Também podemos salvar e carregar os grafos em formatos como: Parquet, JSON e CSV.

GraphFrames are available as an add-on component to GraphX from the Spark Packages website. This Cloudera blog post shows how to use GraphFrames to calculate the PageRank for each node in a graph dataset.

Os GraphFrames estão disponíveis como um componente complementar do GraphX no site Spark Packages. Este post do blog da Cloudera mostra como usar GraphFrames para calcular o PageRank para cada nó em um dataset de grafos.

Algoritmos para analisar grafos

Os algoritmos de grafos permitem executar análises em um dataset de grafos sem que precisemos escrever nossas próprias implementações desses algoritmos. A seguir está uma lista dos algoritmos que ajudam a encontrar padrões em grafos:

• PageRank;
• componentes conectados;
• propagação de labels;
• SVD++;
• componentes fortemente conectados;
• contagem de triângulos;
• caminhos mais curtos:
• detecção de comunidade.

O Spark GraphX vem com um conjunto de algoritmos de grafos pré-construídos para ajudar no processamento e análise dos grafos. Esses algoritmos estão disponíveis no pacote org.apache.spark.graphx.lib. É tão simples quanto chamar esses algoritmos como métodos na classe Graph.

A Figura 2 mostra como os diferentes algoritmos de grafo são construídos sobre a base da API GraphX.

Figura 2. Algoritmos de grafo da biblioteca Spark GraphX.

Vamos examinar mais profundamente os algoritmos: PageRank, componentes conectados e os algoritmos de contagem de triângulos.

PageRank

O algoritmo PageRank é usado para determinar a importância relativa de um objeto dentro de um dataset de grafos. Ele mede a importância de cada nó em um grafo, assumindo que uma aresta de outro nó para este nó representa um endosso.

A engine de pesquisa do Google é um exemplo clássico de PageRank. O Google usa o PageRank para contar quantas outras páginas da web estão vinculadas a uma página da web de destino e usa o resultado como uma medida para determinar a importância dessa página de destino.

Outro exemplo é uma rede social como o Twitter. Um usuário do Twitter seguido por muitos usuários tem um PageRank mais alto do que usuários que não têm tantos seguidores.

O GraphX fornece duas implementações de PageRank: estática e dinâmica. O PageRank estático é executado por um número fixo de iterações para gerar valores de PageRank para um determinado conjunto de nós em um dataset de grafos. O PageRank dinâmico, por outro lado, é executado até que os valores do PageRank convergem com base em um valor de tolerância predefinido.

Componentes conectados

Um componente conectado em um grafo é definido como um subgrafo de nós (ou vértices) que se conectam entre si e nenhum outro nó no grafo principal maior. Um componente conectado é isolado de todos os outros componentes conectados no grafo principal. Isso significa que quaisquer dois nós que pertençam ao mesmo componente conectado devem compartilhar um relacionamento. O menor número de ID entre os nós em um subgrafo é usado para rotular o componente conectado ao qual pertence. Os componentes conectados podem ser usados para criar clusters no grafo - por exemplo, em uma rede social.

Existem duas maneiras de percorrer o grafo para calcular os componentes conectados: com uma pesquisa em largura (breadth-first search - BFS) ou uma pesquisa em profundidade (depth-first search - DFS).

Existe outro algoritmo denominado componentes fortemente conectados (strongly connected components - SCC) para o processamento de grafos. Se todos os nós de um grafo são alcançáveis de todos os outros nós, consideramos o grafo fortemente conectado.

Contagem de triângulos

A contagem de triângulos é um algoritmo de grafo usado para detecção de comunidade que determina o número de triângulos que passam por cada vértice no dataset do grafo. Como o nome indica, um triângulo é um subgrafo de três nós com cada nó conectado aos outros dois. Este algoritmo retorna um grafo e extraímos os vértices deste grafo.

A contagem de triângulos é muito usada na análise de redes sociais. Ele fornece uma medida de agrupamento dos grafos, que é útil para localizar comunidades e medir a coesão das comunidades locais em redes sociais como LinkedIn ou Facebook. O coeficiente de agrupamento, é uma métrica importante em redes sociais, mostra quão firmemente uma comunidade se conecta ou se agrupa em torno de um de seus nós.

Observe que o PageRank é uma medida de relevância, enquanto a contagem de triângulos é uma medida de agrupamento.

Outros casos de uso para o algoritmo de contagem de triângulos são a detecção de spam e as recomendações de links.

A contagem de triângulos é um algoritmo pesado e computacionalmente custoso em comparação com outros algoritmos de grafo, portanto, execute o Spark em uma infraestrutura de hardware decente.

Aplicação de exemplo

Vimos o que são grafos e por que a análise de grafo é uma parte importante nos projetos de processamento de Big Data. Vejamos agora uma aplicação de exemplo que usa alguns dos algoritmos de grafo.

Usaremos datasets das redes sociais Facebook, LiveJournal e YouTube. Todos eles produzem dados conectados e são excelentes recursos para programas de análise de grafos.

Os exemplos que usamos aqui são baseados nos exemplos do GraphX de uma comparação entre o Dato e GraphX.

A versão mais recente dos exemplos apresentados nesta seção estão disponíveis no seguinte projeto Github:

https://github.com/spenchikala/big-data-processing-spark-mini-book

Casos de uso

O principal objetivo dos casos de uso neste exemplo é determinar estatísticas dos grafos, como:

• A popularidade de diferentes usuários na rede social (PageRank);
• Clusters de usuários com base em como os usuários se conectam na rede social (componentes conectados);
• Detecção da comunidade e coesão das comunidades de usuários na rede social (contagem triangular).

Datasets

Nos códigos de exemplo, executaremos algumas análises em datasets diferentes por meio do Spark GraphX. Esses datasets estão disponíveis no site do Stanford Network Analysis Project (SNAP). Para usar esses datasets, baixe-os e copie-os para uma pasta de dados no diretório principal da aplicação de exemplo.

Algoritmos

Usaremos os seguintes algoritmos:

• PageRank nos dados do YouTube;
• Componentes conectados nos dados do LiveJournal;
• Contagem de triângulos nos dados do Facebook.

A Tabela 2 mostra os casos de uso, datasets e algoritmos usados no processamento dos grafos.

Tabela 2: Datasets e algoritmos usados na aplicação de exemplo do Spark GraphX.

Tecnologias

Usaremos as seguintes tecnologias na aplicação de exemplo de análise de grafo.

Tabela 3. Tecnologias e ferramentas usadas na aplicação de exemplo.

Código do exemplo

Vamos escrever o código usando a linguagem de programação Scala. Usaremos a ferramenta de linha de comando do Spark shell para executar esses programas. Esta é a maneira mais rápida de verificar os resultados do programa. Não precisamos de nenhuma compilação de código adicional ou etapas de construção.

Esses programas estão disponíveis como um arquivo .zip para você baixar e testar em seu próprio ambiente de desenvolvimento.

Vejamos os detalhes de cada um dos códigos de exemplo do GraphX.

Primeiro, executaremos o PageRank nos dados da rede social do YouTube. Este dataset inclui comunidades de confiança, que são basicamente grupos definidos pelos usuários que os outros usuários podem ingressar.

PageRank:

import org.apache.spark._
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
import java.util.Calendar

// Carrega os vértices de um grafo.
val graph = GraphLoader.edgeListFile(sc, "data/page-rank-yt-data.txt")
// Pega os detalhes do grafo como arestas, vértices, etc.

val vertexCount = graph.numVertices

val vertices = graph.vertices
vertices.count()

val edgeCount = graph.numEdges

val edges = graph.edges edges.count()

val triplets = graph.triplets
// O método collect() demora um bom tempo para executar.
// triplets.collect()
triplets.count()
triplets.take(5)

val inDegrees = graph.inDegrees
inDegrees.collect()

val outDegrees = graph.outDegrees
outDegrees.collect()

val degrees = graph.degrees
degrees.collect()
			
// Quantidade de iterações passado via argumento.
val staticPageRank = graph.staticPageRank(10)
staticPageRank.vertices.collect()

Calendar.getInstance().getTime()
val pageRank = graph.pageRank(0.001).vertices
Calendar.getInstance().getTime()

// Imprime os 5 primeiros itens do resultado.
println(pageRank.top(5).mkString("\n"))

Vejamos agora o código que executa os componentes conectados nos dados da rede social do LiveJournal. Este dataset inclui usuários registrados que enviaram posts de blog individuais e em grupo. O site LiveJournal também permite que os usuários identifiquem outros usuários como amigos.

Componentes conectados:

import org.apache.spark._
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
import java.util.Calendar
					
// Componentes conectados.
val graph = GraphLoader.edgeListFile(sc, "data/connect ed-components-lj-data.txt")

Calendar.getInstance().getTime()
val cc = graph.connectedComponents()
Calendar.getInstance().getTime()

cc.vertices.collect()
// Imprime os primeiros 5 itens do resultado.
println(cc.vertices.take(5).mkString("\n"))
val scc = graph.stronglyConnectedComponents()

scc.vertices.collect()

Finalmente, este programa Spark, que realiza a contagem de triângulos nos dados de círculos sociais do Facebook. O dataset inclui as listas de amigos do Facebook com perfis de usuário, círculos e redes de ego.

Contagem de triângulos:

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD

val graph = GraphLoader.edgeListFile(sc, "data/trian gle-count-fb-data.txt")

println("Quantidade de vértices: " + graph.vertices. count())
println("Quantidade de arestas: " + graph.edges.count())

graph.vertices.foreach(v => println(v))
val tc = graph.triangleCount()
tc.vertices.collect
println("tc: " + tc.vertices.take(5).mkString("\n"));

// println("Contagem de triângulos: " + graph.connectedComponents.triangleCount().vertices.collect().mkString("\n"));
println("Contagem de triângulos: " + graph.connectedComponents.triangleCount().vertices.top(5).mkString("\n"));

val sum = tc.vertices.map(a => a._2).reduce((a, b) => a + b)

Conclusões

Com o crescimento cada vez maior dos dados conectados nas empresas, agências governamentais e empresas de mídia social, o processamento e análise de grafos se tornarão mais críticos em análises preditivas e soluções de engine de recomendação à medida que fornecem insights e serviços para os funcionários, clientes e usuários.

O Spark GraphX ​​é uma excelente escolha para o processamento de grafos. Ele fornece um algoritmo de processamento de dados unificado e um conjunto de ferramentas para fornecer insights valiosos e modelos de previsão sobre os dados conectados gerados pelos processos de negócios nas organizações.

O que vem a seguir?

Como vimos, o framework do Apache Spark fornece as bibliotecas, utilitários e ferramentas necessárias para arquiteturas de aplicações unificadas de processamento de Big Data. Quer precisemos processar os dados em tempo real, em batch ou se o dataset tiver conexões e relacionamentos, o Spark facilita o trabalho com diferentes tipos de dados. Não precisamos mais depender de vários frameworks diferentes para processar e analisar os diferentes tipos de dados gerados em uma organização típica.

Se estamos procurando uma solução de Big Data para usar nas aplicações em nossas organizações ou estamos interessados ​​em aprender sobre Big Data e ciência de dados, o Spark é um excelente framework para aprender e usar nas aplicações.

Referências

• Apache Spark homepage
• Apache Spark GraphX homepage
• GraphX Programming Guide
• Spark GraphX in Action (Manning Publications)
• "Facebook's Comparison of Apache Giraph and Spark GraphX for Graph Data Processing"
• Post sobre GraphFrames no blog da Databricks

Sobre o autor

Srini Penchikala atualmente trabalha como Arquiteto de Software em uma organização de serviços financeiros em Austin, Texas. Ele tem mais de 20 anos de experiência em arquitetura, design e desenvolvimento. Srini está atualmente escrevendo um livro sobre Apache Spark. Ele também é co-autor do livro "Spring Roo in Action" da Manning Publications. Ele palestrou em diversas conferências como: JavaOne, SEI Architecture Technology Conference (SATURN), IT Architect Conference (ITARC), No Fluff Just Stuff, NoSQL Now e Project World Conference. Srini também publicou diversos artigos sobre arquitetura de software, segurança e gerenciamento de risco, e sobre banco de dados NoSQL em sites como o InfoQ, The ServerSide, OReilly Network (ONJava), DevX Java, java.net e JavaWorld. Ele é o Editor Líder de Data Science na comunidade do InfoQ.

Este é o sexto artigo da série "Big Data com Apache Spark". Por favor, veja também a Parte 1: Introdução, Parte 2: Spark SQL, Parte 3: Spark Streaming, Parte 4: Spark Machine Learning, Parte 5: Spark ML Data Pipelines.