Big Data com Apache Spark Part 3: Spark Streaming

Este é o terceiro artigo da série Big Data com Apache Spark. Nos 2 primeiros artigos, Introdução e Spark SQL, abordamos o processamento de dados estáticos.

As soluções descritas nos artigos anteriores se basearam no processamento de dados estáticos por meio de processamento em lote, por exemplo como uma atividade diária ou que é executada em um horário específico e predeterminado. Mas o que acontece com dados que precisam ser processados em tempo real e que tem o objetivo de gerar insights para tomada de decisões em negócios?

Para dados em streaming - dados gerados continuamente - o processamento precisa ser realizado em tempo real à medida que os dados chegam e não em um processo em lote. O processamento e a análise de dados em tempo real está se tornando um componente crítico na estratégia de Big Data para a maioria das empresas e negócios.

Neste artigo examinaremos um caso de uso sobre dados históricos de arquivos em log de um servidor Web com o objetivo de mostrar como o Spark Streaming pode ajudar com a execução da análise de dados que são gerados de forma contínua.

Análise de dados em Streaming

Dados em streaming basicamente dizem respeito a um agrupamento de registros de informações gerados a partir de fontes como sensores, tráfego de servidores e pesquisas online. Para alguns exemplos de dados em streaming podemos citar a atividade dos usuários em um site, os dados de monitoramento, os logs do servidor e qualquer outro tipo de dados relacionados a eventos.

Aplicações destinadas ao processamento de dados em streaming auxiliam a análise por meio de dashboards, recomendações online e em tempo real ou por exemplo na detecção de fraude em tempo real.

Ao criar aplicações para coletar, processar e analisar dados em streaming é necessário levar em consideração uma abordagem diferente de uma abordagem utilizada para tratar dados que normalmente são processados em lote.

Existem diferentes frameworks no mercado voltados ao processamento de dados em streaming. Entre eles podemos destacar:

• Apache Samza
• Storm
• Spark Streaming

O Spark Streaming

O Spark Streaming é uma extensão que faz parte do núcleo da API Spark. O Spark Streaming facilita a criação de fluxos de processamento tolerante a falhas sobre dados em streaming e em tempo real.

A figura 1 abaixo mostra como o Spark Streaming se encaixa no ecossistema geral do Apache Spark.

Figura 1 - Ecossistema Spark com a biblioteca Spark Streaming

A forma como o Spark Streaming funciona é dividindo o streaming em lotes (chamados micro lotes) em um intervalo predefinido (N segundos) e, em seguida, trata cada lote de dados como arquivos Resilient Distributed Datasets (RDDs). Com isto é possível processar esses RDDs por meio de operações como map, reduce, reduceByKey, join e window. Os resultados dessas operações RDD são retornados em lotes. Normalmente, estes resultados são armazenados em um banco de dados para futuras análises e também com o objetivo de gerar relatórios e dashboards ou mesmo enviar alertas baseados no eventos interpretados a partir das informações coletadas.

É importante decidir o intervalo de tempo a ser utilizado pelo Spark Streaming, baseado no caso de uso e nos requisitos de processamento de dados o qual se quer analisar. Se o valor de N é muito baixo, então os micro lotes não terão dados suficientes para dar resultados significativos durante a análise.

Comparado ao Spark Streaming, outros frameworks para processamento de streaming processam streamings se baseando em eventos e não como micro lotes. Com esta abordagem em micro lotes, é possível utilizar outras bibliotecas Spark (como Core, Machine Learning, etc...) com a API do Spark Streaming em uma mesma aplicação.

Os dados em streaming podem ser originados em fontes de informação diferentes. Algumas dessas fontes de informação incluem:

• Kafka
• Flume
• Twitter
• ZeroMQ
• Amazon's Kinesis
• TCP sockets

Outra vantagem de usar um framework para processamento de Big Data como o Apache Spark é que podemos combinar processamento em lote e processamento de streaming em um mesmo sistema. Também é possível aplicar algoritmos de aprendizado de máquina e para processamento de grafos do Spark nestes dados em streaming. Discutiremos as bibliotecas de Aprendizado de Máquina e de Processamento de Grafos MLlib e GraphX respectivamente, em artigos futuros desta série.

A arquitetura da biblioteca Spark Streaming é apresentada na figura 2 abaixo:

(Clique na imagem para ampliar)

Figura 2 - Como p Spark Streaming funciona

Casos de uso conhecidos com o Spark Streaming

O Spark Streaming está se tornando a plataforma escolhida para implementar soluções de processamento de dados e análise de dados em tempo real de informações originadas em aplicações com a Internet das Coisas (IoT) e em sensores. Esta biblioteca integrante da família Apache Spark vem sendo utilizado em uma variedade de casos de uso e aplicações de negócios.

Alguns dos casos de uso mais interessantes do Spark Streaming incluem:

• O Uber, a empresa por trás do serviço de compartilhamento de caronas, usa o Spark Streaming em seu pipeline Streaming de ETL contínuo para coletar terabytes de dados de eventos todos os dias sobre seus usuários móveis e com isto, é capaz de fazer análise de telemetria em tempo real.
• A Pinterest, a empresa por trás da ferramenta de bookmarking visual, utiliza as tecnologias Spark Streaming, MemSQL e Apache Kafka para fornecer insights sobre como seus usuários estão envolvidos com os Pins em todo o mundo e em tempo real.
• A Netflix usa o Kafka e o Spark Streaming para construir recomendações de filmes online e uma solução de monitoramento de dados em tempo real que processa bilhões de eventos recebidos por dia de diferentes fontes de dados.

Outros exemplos do mundo real com Spark Streaming incluem:

• Análise da cadeia de suprimentos (Supply chain analytics)
• Operações de segurança e inteligência em tempo real para localizar ameaças
• Plataforma de anúncio de leilões
• Análise de vídeo em tempo real para auxiliar com as experiências personalizadas e interativas aos espectadores

Vamos detalhar a arquitetura do Spark Streaming e os métodos de sua API. Para escrever programas no Spark Streaming, há dois componentes que são necessário conhecer: o DStream e o StreamingContext.

DStream

DStream (abreviação para Discretized Stream) é uma abstração básica no Spark Streaming e representa um fluxo de dados contínuo. DStreams podem ser criados a partir de dados de entrada originados em fontes de informação como o Kafka, o Flume e o Kinesis, ou aplicando operações em outros DStreams. Internamente, um DStream é representado como uma seqüência de objetos RDD.

Semelhante às operações de transformação e ação em RDDs, os DStreams suportam as seguintes operações:

• map
• flatMap
• filter
• count
• reduce
• countByValue
• reduceByKey
• join
• updateStateByKey

Streaming Context

Semelhante ao SparkContext no Spark, StreamingContext é o ponto de entrada principal para toda a funcionalidade de streaming.

O StreamingContext possui métodos internos para receber dados gerados continuamente em uma aplicação Spark Streaming.

Usando esse contexto, podemos criar um DStream que represente os dados de um streaming em uma origem TCP que esteja especificada como por exemplo um hostname e o número de uma porta. Por exemplo, se estivermos usando uma ferramenta como o netcat para testar um programa Spark Streaming, receberíamos as informações da máquina onde o netcat está sendo executado (por exemplo, localhost) e um número de porta de 9999.

Quando o código é executado, o Spark Streaming só configura o processamento que deverá ser executado quando ele for de fato iniciado, e nenhum processamento real é feito até este momento. Para iniciar o processamento após todas as transformações estarem configuradas, o método start () é acionado e o método awaitTermination() é utilizado para aguardar o encerramento do processamento.

A API do Spark Streaming

O Spark Streaming vem com vários métodos em sua API que são úteis para processar dados em streaming. Existem operações semelhantes às operações com RDD como por exemplo map, flatMap, filter, count, reduce, groupByKey, reduceByKey, sortByKey e join. Ele também fornece uma API adicional para processar dados em streaming baseado em janelas e operações stateful. Estes métodos adicionais incluem window, countByWindow, reduceByWindow, countByValueAndWindow, reduceByKeyAndWindow e updateStateByKey.

A biblioteca Spark Streaming é atualmente suportada em linguagens de programação como Scala, Java e Python. Abaixo destacamos os links para a API do Spark Streaming em cada uma destas linguagens de programação.

• Spark Streaming Scala API
• Java API
• Python API

Passos em um programa Spark Streaming

Antes de iniciar a discussão em torno de uma aplicação de exemplo,, vamos dar uma olhada nas diferentes etapas envolvidas em um programa típico em Spark Streaming.

• O Spark Streaming Context é usado para processar streams de dados em tempo real. Assim, a primeira etapa é inicializar o objeto StreamingContext usando dois parâmetros, SparkContext e tempo de intervalo de deslocamento. O intervalo de deslocamento define a janela de atualização onde processamos os dados que chegam como streams. Uma vez que o contexto é inicializado, nenhum novo processamento pode ser definido ou adicionado ao contexto existente. Além disso, apenas um objeto StreamingContext pode estar ativo ao mesmo tempo.
• Uma vez que o contexto é definido no Spark Streaming, especificamos as fontes de dados de entrada criando DStreams de entrada. Em nosso aplicativo de exemplo, a fonte de dados de entrada é um gerador de mensagens de log que usa o banco de dados distribuído Apache Kafka e o sistema de mensagens do sistema operacional (syslog no Linux). O programa gerador de log cria mensagens de log aleatórias para simular um ambiente de execução em tempo real de um servidor web onde as mensagens de log são continuamente geradas, pois vários aplicativos da Web atendem ao tráfego de usuários.
• Define os processamentos usando a API de transformações do Sparking Streaming como um mapa e utiliza o método reduce para os DStreams.
• Depois que a lógica de processamento do stream é definida, podemos começar a receber os dados e processá-los usando o método start no objeto StreamingContext criado anteriormente.
• Finalmente, aguardamos que o processamento dos dados em stream seja concluído usando o método awaitTermination do objeto StreamingContext.

Aplicação de Exemplo

A aplicação de exemplo que será discutida neste artigo é um programa de processamento e análise de log de servidor. Ele pode ser usado para monitorar em tempo real os logs de um servidor de aplicativos e realizar análises de dados nesses logs. Essas mensagens de log são consideradas dados de séries temporais, que é definido como uma seqüência de pontos de dados contendo medidas sucessivas capturadas em um intervalo de tempo especificado.

Exemplos de dados de séries temporais incluem dados de sensores, informações meteorológicas e cliques de dados em stream. A análise sobre séries de dados temporais trata do processamento destes dados com o objetivo de extrair insights que podem ser usados para a tomada de decisões de negócios. Esses dados também podem ser usados para análise preditiva com o objetivo de prever valores futuros com base em dados históricos.

Com uma solução como essa, não precisamos de um job em lotes executado por hora ou diariamente para processar os logs do servidor. O Spark Streaming recebe dados gerados continuamente, processa estes dados e calcula as estatísticas de log para fornecer informações sobre os dados.

Para seguir um exemplo padrão na análise dos logs de um servidor, usaremos o Apache Log Analyzer discutido na Aplicação de Referência para Spark Streaming da Data Bricks como referência para a nossa aplicação de exemplo. Esta aplicação já possui um código de análise de mensagens de log que vamos reutilizar em nossa aplicação. A aplicação de referência é um excelente recurso para aprender mais sobre o Spark framework em geral e o Spark Streaming em particular. Para obter mais detalhes sobre a Aplicação de Referência em Spark da Data Bricks acesse seu website.

Caso de Uso

Na aplicação de exemplo, analisamos os logs do servidor web para calcular as seguintes estatísticas para análise de dados futuros e também criar relatórios e dashboards:

• A contagem da resposta com os códigos de resposta para diferentes requisições HTTP;
• O tamanho do conteúdo da resposta;
• O endereço IP dos clientes para avaliar qual a origem do maior tráfego Web entrante;
• Identificar as principais URLs de acesso direto com o objetivo de identificar quais serviços são acessados mais com relação aos demais.

Ao contrário dos dois artigos anteriores desta série, vamos usar Java no lugar de Scala para criar o programa Spark deste artigo. Também executaremos o programa como um aplicativo autônomo em vez de executar o código a partir da janela do console. É assim que implementaríamos os programas Spark em ambientes de teste e produção. A interface do console Shell (usando linguagens Scala, Python ou R) é somente para testes de desenvolvedores locais.

Tecnologias

Usaremos as seguintes tecnologias na aplicação exemplo com a finalidade de demonstrar como a biblioteca Spark Streaming é usada para processar streams de dados em tempo real.

Zookeper

Zookeeper é um serviço centralizado que fornece coordenação distribuída e de forma confiável para aplicações distribuídas. O Kafka, o sistema de mensagens que usamos na aplicação exemplo, depende do Zookeeper para os detalhes de configuração entre o cluster.

Apache Kafka

O Apache Kafka é um sistema de mensagens em tempo real, tolerante a falhas, e também é um sistema escalável para mover dados em tempo real. É um bom candidato para casos de uso, como capturar a atividade do usuário em sites, logs, dados de inventário e dados de instrumentação.

O Kafka funciona como um banco de dados distribuído e é baseado em um log com commit de baixa latência particionado e replicado. Quando enviamos uma mensagem para o Kafka, ela é replicada para diferentes servidores no cluster e, ao mesmo tempo, também é persistida - commit - em disco.

O Apache Kafka inclui uma API cliente, bem como um frameworks para transferência de dados batizado de Kafka Connect.

Clientes Kafka: O Kafka inclui clientes Java (tanto para os geradores de mensagens como para os consumidores). Usaremos a API do cliente de um gerador de mensagens em Java para atender nossa aplicação exemplo.

Kafka Connect: O Kafka também inclui o Kafka Connect, que trata-se de um framework para streaming de dados entre o Apache Kafka e sistemas de dados externos capaz de suportar os pipelines de dados nas empresas. Suas funcionalidades incluem a importação e exportação de conectores para mover conjuntos de dados para dentro e para fora do Kafka. O programa Kafka Connect pode ser executado como um processo autônomo ou como um serviço distribuído e suporta uma interface REST para enviar os conectores ao cluster Kafka Connect usando uma API REST.

O Spark Streaming

Neste artigo, faremos uso da API Java do Spark Streaming para receber os streams de dados, calcular as estatísticas de log e executar consultas para responder a perguntas como por exemplo, quais são os endereços IP de onde mais solicitações da web estão vindo entre outras perguntas.

A Tabela 1 abaixo mostra as tecnologias, ferramentas e suas versões usadas na aplicação exemplo.

Tabela 1 - Tecnologias e ferramentas utilizadas na aplicação exemplo com Spark streaming

Os diferentes componentes arquiteturais de nossa aplicação com Spark Streaming são ilustrados na Figura 3.

(clique na imagem para ampliar)

Figura 3 - Componentes arquiteturais da aplicação exemplo

Execução da aplicação exemplo no Spark Streaming

Para configurar o projeto Java localmente, você pode fazer o download do código da aplicação de referência da Databricks no Github. Depois de obter o código da aplicação de referência, você precisará de duas classes Java adicionais para executar a aplicação exemplo.

• O gerador de log (SparkStreamingKafkaLogGenerator.java)
• E o analisador de log (SparkStreamingKafkaLogAnalyzer.java)

Esses arquivos são fornecidos em um arquivo zip (spark-streaming-kafka-sample-app.zip) no site deste artigo. Se você quiser executar a aplicação exemplo em sua máquina local, use o link descrito anteriormente para baixar o arquivo zip, extrair as classes Java e adicioná-las ao projeto Java criado na etapa anterior.

A aplicação exemplo pode ser executada em diferentes sistemas operacionais. Fizemos testes com ambientes Windows e Linux (VM CentOS).

Vejamos cada componente na arquitetura da aplicação e as etapas para executar o programa utilizando o Sparking Streaming.

Comandos Zookeeper

Utilizamos o Zookeeper versão 3.4.6 na aplicação exemplo. Para iniciar o servidor, defina duas variáveis de ambiente, JAVA_HOME e ZOOKEEPER_HOME para apontar para os diretórios de instalação do JDK e do Zookeeper, respectivamente. Em seguida, navegue até o diretório inicial do Zookeeper e execute o seguinte comando para iniciar o Zookeeper server.

bin\zkServer.cmd

Caso esteja testando em um ambiente Linux, os comandos são:

bin/zkServer.sh start

Comandos no servidor Kafka

A versão 2.10-0.9.0.0 do Kafka foi utilizada na aplicação exemplo, e esta versão é baseada na versão 2.10 da linguagem Scala. A versão da linguagem Scala que você usa com Kafka é muito importante porque se a versão correta não for utilizada, você obtém erros de execução ao executar o programa do Spark streaming. Abaixo descrevemos os passos para iniciar a instância no servidor Kafka:

• Abra um novo console de terminal
• Configure as variáveis JAVA_HOME e KAFKA_HOME
• Acesse o diretório raiz do Kafka
• Execute o seguinte comando

bin\windows\kafka-server-start.bat config\server.properties

Em ambientes Linux os comandos são os seguintes:

bin/kafka-server-start.sh config/server.properties

Comandos no gerador de logs

O próximo passo em nossa aplicação exemplo é executar o gerador de mensagens de log.

O gerador de log gera mensagens de log de teste com diferentes códigos de resposta HTTP (como 200, 401 e 404) com URLs destino diferentes.

Antes de executar o gerador de log, precisamos criar um tópico o qual possamos escrever as mensagens.

Semelhante a etapa anterior, abra um novo console de linha de comando, defina as variáveis JAVA_HOME e KAFKA_HOME e acesse o diretório raiz do Kafka. Em seguida, execute o seguinte comando inicial para visualizar os tópicos existentes e disponíveis no servidor Kafka.

bin\windows\kafka-run-class.bat kafka.admin.TopicCommand --zookeeper localhost:2181 --list

No linux execute:

bin/kafka-run-class.sh kafka.admin.TopicCommand --zookeeper localhost:2181 --list

Vamos criar um novo tópico nomeado "spark-streaming-sample-topic" usando o seguinte comando:

bin\windows\kafka-run-class.bat kafka.admin.TopicCommand --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --create --topic spark-streaming-sample-topic

Em ambiente Linux execute:

bin/kafka-run-class.sh kafka.admin.TopicCommand --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --create --topic spark-streaming-sample-topic

Podemos executar o comando para listar os tópicos novamente para ver se o novo tópico foi criado corretamente.

Depois que o tópico foi criado, podemos executar o programa gerador de log. Isso é feito executando a classe Java SparkStreamingKafkaLogGenerator. A classe do gerador de log leva os seguintes quatro argumentos para especificar os parâmetros de configuração.

• Group ID: spark-streaming-sample-group
• Topic: spark-streaming-sample-topic
• Número de interações: 50
• Intervalo: 1000

Abra um novo console de linha de comando para executar o gerador de log. Definiremos três variáveis de ambiente (JAVA_HOME, MAVEN_HOME e KAFKA_HOME) para os diretórios JDK, Maven e Kakfa, respectivamente. Em seguida, acesse o diretório raiz do projeto exemplo (por exemplo, c:\dev\projects\spark-streaming-kafka-sample-app) e execute o seguinte comando.

mvn exec:java -Dexec.mainClass=com.sparkstreaming.kafka.example.SparkStreamingKafkaLogGenerator -Dexec.args="spark-streaming-sample-groupid spark-streaming-sample-topic 50 1000"

Uma vez que o programa gerador de log está em execução, você deve ver as mensagens de log de teste criadas juntamante com as mensagens de depuração exeibidas no console. Este é apenas um código de exemplo, portanto, as mensagens de log são geradas aleatoriamente para simular dados em streaming de uma loja de eventos como um servidor da Web.

A Figura 4 apresenta uma captura de tela das mensagens de log bem como as mensagens de log sendo geradas.

(clique na imagem para ampliar)

Figura 4 - Captura de tela da geração de logs na aplicação exemplo

Comandos do Spark Streaming

Este é o consumidor de mensagens de log usando a API do Spark Streaming. Usamos uma classe Java denominada SparkStreamingKafkaLogAnalyzer para receber os fluxos de dados do servidor Kafka e processá-los para criar estatísticas do log.

O Sparking Streaming processa mensagens de log do servidor e gera estatísticas de log cumulativas como o tamanho do conteúdo do pedido da web (mínimo, máximo e médio), a contagem dos códigos de resposta, os endereços IP e as principais URLs diretamente acessadas.

Criamos o Spark Context utilizando o parâmetro "local [*]", que detecta o número de núcleos no sistema local e os usa para executar o programa.

Para executar a classe Java Spark Streaming, será necessário os seguintes arquivos JAR no classpath:

• kafka_2.10-0.9.0.0.jar
• kafka-clients-0.9.0.0.jar
• metrics-core-2.2.0.jar
• spark-streaming-kafka_2.10-1.4.0.jar
• zkclient-0.3.jar

Executei o programa no Eclipse IDE depois de adicionar os arquivos JAR descritos acima ao classpath. A saída do programa de análise de log no Streaming é apresentada na Figura 5.

(clique na imagem para aumentar)

Figura 5 - Saída do programa de análise de log no Spark Streaming

Quando o Spark Streaming está em execução podemos verificar o console do Spark e com isto visualizar os detalhes do processamento dos jobs no Spark.

Abra uma nova aba do navegador web e acesse a URL http://localhost:4040 para acessar o console do Spark.

Vejamos alguns dos gráficos que mostram as estatísticas do programa no Spark Streaming.

A primeira visualização diz respeito ao DAG (Direct Acyclic Graph) de um job específico mostrando o gráfico de dependência de diferentes operações que executamos no programa, como map, window e foreachRDD. A Figura 6 abaixo mostra a captura de tela desta visualização do job no Spark Streaming em nosso programa exemplo.

(clique na imagem para ampliar)

Figura 6 - Gráfico de visualização DAG de um job em Spark Streaming

O próximo gráfico que podemos observar é relativo às estatísticas de streaming que incluem a taxa de entrada mostrando o número de eventos por segundo e o tempo de processamento em milissegundos.

A Figura 7 mostra essas estatísticas durante a execução do programa Spark Streaming quando os dados de streaming não estão sendo gerados (seção esquerda) e quando o fluxo de dados está sendo enviado para o Kafka e processado pelo consumidor do Spark Streaming (seção à direita).

(clique na imagem para aumentar)

Figura 7 - Visualização das estatística do streaming no Spark Streaming para o programa exemplo

Conclusão

A biblioteca Spark Streaming, parte do ecossistema Apache Spark, é usada para processamento de streaming de dados em tempo real. Neste artigo, aprendemos como usar a API do Spark Streaming para processar dados gerados pelos logs de um servidor e executar análises nestes streaming de dados em tempo real.

Próximos passos

O aprendizado de máquina, a análise preditiva e a ciência de dados estão recebendo muita atenção recentemente para resolver problemas em diferentes casos de uso. A biblioteca Spark MLlib e a biblioteca de aprendizado de máquina do Spark fornecem vários métodos internos para usar diferentes algoritmos de aprendizagem de máquina como filtragem Colaborativa, agrupamento e classificação.

No próximo artigo, vamos explorar a biblioteca Spark MLlib e explorar dois casos de uso para ilustrar como podemos aproveitar as capacidades de ciência de dados do Spark para tornar mais fácil executar algoritmos de aprendizado de máquina.

Nos próximos artigos desta série veremos frameworks como BlinkDB e Tachyon.

Referências

• Big Data com Apache Spark - Parte 1: Introdução
• Big Data com Apache Spark - Parte 2: Spark SQL
• Site oficial do Apache Spark
• Website do Spark Streaming
• Spark Streaming Guia de Programação
• Spark Streaming - Exemplo de códigos em Scala
• Spark Streaming - Exemplos de código em Java
• Aplicação de Referência da Data Bricks' utilizando Apache Spark
• Marcando e Processando Dados em Tempo Real utilizando Spark Streaming - Spark Summit 2015 Conference
• Guia de Referência do MapR's utilizando Spark Streaming

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 padrões de bancos de dados NoSQL e 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 banco de dados NoSQL na comunidade do InfoQ.