Monitorar performance de aplicações Spark usando o profiler de JVM do Uber, com InfluxDB e Grafana

Muitas indústrias estão usando o Apache Spark para construção de aplicações de processamento de grandes volumes de dados. O Spark provê paralelismo de dados implícito e tolerância à falhas para este tipo de aplicação. A aplicação pode ser para processamento de streaming, processamento em lote, processamento de conjunto de dados SQL ou aplicação de aprendizado de máquina. O motor de processamento de dados rápido e em memória do Spark executa as aplicações em um cluster e transforma e processa uma grande quantidade de dados em um pipeline de dados. A execução contínua e confiável dessas aplicações é uma tarefa desafiadora e um bom sistema de monitoramento de desempenho é necessário. Já que o Spark está sendo amplamente adotado pelas indústrias, o monitoramento de desempenho, a análise de métricas e a correção de problemas das aplicações Spark estão recebendo mais atenção. O Uber recentemente liberou o código fonte do seu profiler de JVM para Spark. Neste artigo, será discutido como estender o profiler de JVM do Uber e usá-lo com o InfluxDB e o Grafana para monitorar e reportar as métricas de desempenho de uma aplicação Spark.

Sistema de monitoramento de performance de aplicações Spark

Um sistema de monitoramento de desempenho é necessário para a utilização ideal dos recursos disponíveis e a detecção antecipada de possíveis problemas. Ele deve fornecer relatórios de fácil compreensão de status dos sistemas em execução e enviar alertas sobre falhas de componentes. Quando falamos em sistemas distribuídos em larga escala em execução num cluster Spark em conjunto com diferentes componentes do ecossistema Hadoop, a necessidade de um sistema de monitoramento de desempenho refinado torna-se predominante. A aplicação Spark executa o processamento distribuído de dados e opera sobre recursos compartilhados, o que torna o trabalho da equipe de DevOps muito complexo. A equipe de DevOps precisa gerenciar os recursos disponíveis de forma eficaz e monitorar de perto os diferentes componentes do sistema para evitar qualquer indisponibilidade. A visibilidade da pilha completa fornecida por um sistema de monitoramento de desempenho efetivo ajuda a equipe de DevOps a entender o comportamento do sistema e a reagir rapidamente aos problemas em ambiente de produção. Isto garante a confiabilidade, escalabilidade e o desempenho da aplicação Spark.

Um sistema de monitoramento de performance ideal que atenda um sistema complexo precisa ter as seguintes características:

• O sistema de monitoramento deve fornecer visibilidade granular sobre cada componente dentro do cluster. Devemos obter métricas detalhadas sobre CPU, memória, armazenamento, E/S (Entrada e Saída) de disco para arquivos locais e HDFS, rastreamentos de pilha, etc. Essas métricas ajudam a solucionar rapidamente as instâncias com problemas.
• O sistema de monitoramento deve fornecer métricas de nível de código para as aplicações (por exemplo, tempo de execução, argumentos usados por diferentes métodos) em execução no Spark. Isto ajudará a identificar métodos de execução lentos, discos com alta taxa de leitura/escrita, etc.
• O sistema de monitoramento deve armazenar métricas para cada segundo e permitir analisá-las navegando por diferentes períodos de tempo. Deve-se permitir a análise dos dados em segundo e em terceiro nível. Deve-se permitir controlar o período de retenção dos dados e facilmente acessar e analisar os dados passados quando necessário. Isto ajuda analisar a tendência atual e prever tendências futuras.
• O sistema de monitoramento deve fornecer uma maneira eficaz de extrair informações significativas de um grande volume de métricas sendo que são coletadas continuamente. Isso inclui consulta de dados usando SQL ou API, filtragem de dados, agregação de valores e aplicação de análise personalizada. Isso ajuda na transformação e na análise mais rápida de dados.
• O sistema de monitoramento deve fornecer acesso fácil a informações resultantes da análise de dados de métricas. Deve haver a opção de exibir os dados em painel usando formas diferentes, como gráficos, tabelas de gráficos, etc. Deve haver a opção de categorizar os dados com base no host, horário ou trabalho, e os usuários devem ser capazes de detalhá-los para correlacionar os diferentes pontos de dados. Deve haver uma provisão para configurar alertas e notificações para limites definidos pelo usuário em dados de métricas. Isso ajuda as equipes de DevOps e outros envolvidos de uma organização a obter rapidamente o conjunto de informações desejado quando necessário.

Neste artigo, será desenvolvido um sistema de monitoramento de desempenho usando ferramentas e tecnologias de código aberto. O sistema de monitoramento de desempenho de uma aplicação Spark é projetado com três objetivos:

1. Coletar as métricas de desempenho do sistema (guia e executores) e código da aplicação;
2. Armazenar as métricas de forma durável para análise de série temporal (lote e tempo-real);
3. Gerar relatórios de dados de métricas na forma de gráficos e tabelas.

O Apache Spark fornece uma interface web e uma API REST para métricas. O Spark também fornece uma variedade de dissipadores, incluindo consoles, JMX, Servlet, Grafite, etc. Há algumas outras ferramentas de monitoramento de desempenho de software livre disponíveis, como dr-elephant, sparklint, prometheus, etc. As métricas fornecidas por essas ferramentas são principalmente métricas no nível do servidor e algumas delas também fornecem informações sobre a execução de aplicações.

O profiler de JVM do Uber coleta as métricas tanto no nível do servidor quanto de código da aplicação. Esse profiler pode coletar todas as métricas (cpu, memória, buffer pool, etc) do guia, executor ou qualquer JVM. Ele pode instrumentar código existente sem modificá-lo, para que possa coletar métricas sobre métodos, argumentos e tempo de execução. Para armazenar métricas para análise de série temporal, usaremos o InfluxDB, que é um poderoso banco de dados de séries temporais. Vamos estender o Uber JVM Profiler e adicionar um novo gerador de relatório para o InfluxDB para que os dados das métricas possam ser armazenados usando a API HTTP. Para o painel de gráficos, usaremos o Grafana, que consultará o InfluxDB para dados de métricas.

Abaixo estão os detalhes sobre as ferramentas/ tecnologias usadas para o sistema de monitoramento de desempenho de aplicações Spark.

Profiler de JVM do Uber

O profiler de JVM do Uber é um profiler distribuído que coleta métricas de desempenho e métricas de utilização de recursos de diferentes nós num cluster. Esse profiler JVM é executado como um agente Java junto com a aplicação e coleta diferentes métricas. Ele publica essas métricas para um gerador de relatório designado para análises e relatórios adicionais. O profiler de JVM do Uber foi desenvolvido para criação de perfis de aplicações Spark, mas pode ser usado para criar perfis de qualquer aplicação baseada na JVM. Existem três componentes principais do profiler de JVM do Uber:

1. Profiler: Profiler de JVM do Uber vem com os seguintes profilers integrados:

• Processador e Memória - Coleta de espaço de contingência (direto e mapeado), coleta de lixo (contador e tempo), Memória Heap (comprometida e total usada), Memória Não-Heap (comprometida e total usada), CPU (carga e tempo), detalhes do espaço de memória (EdenSpace, SurvivorSpace, TenuredGen, CodeCache, CompressedClassSpace, Metaspace), métricas vmHWM e vmRSS.
• E/S (Entrada e Saída) - Coleta de estatísticas da CPU (ócio, estado, usuário, tempo de espera de E/S) e bytes de E/S de leitura e escrita de disco.
• Rastreamento de pilha - Coleta os nomes das estruturas dos processos, estado das estruturas dos processos e métricas de rastreio das pilhas.
• Informação de processo - Coleta a versão do agente, caminho da classe JVM, argumentos de entrada da JVM, e métricas xmxBytes.
• Duração de método - Coleta métricas do tempo de execução de um método com o nome da classe, nome do método e nome do processo.
• Argumento de método - Coleta métricas dos argumentos dos métodos com nome da classe, nome do método e nome do processo.

2. Transformador: Este é um Transformador de Arquivo de Classe para instrumentar códigos em bytes de métodos Java. Este transformador é usado para instrumentar métodos dos profilers Duração de Método e Argumento do Método.

3. Repórter: Os seguintes repórteres estão disponíveis:

• Saída para o Kafka - para enviar métricas de desempenho à um tópico do Kafka.
• Saída para um arquivo - para escrever métricas em um arquivo.
• Saída para o console - para enviar métricas ao console.
• Saída para o Redis - para armazenar métricas em banco de dados Redis.

Para mais detalhes sobre o profiler de JVM por favor consulte o publicação no blog do Uber.

InfluxDB e Grafana

InfluxDB: O InfluxDB é um banco de dados de séries temporais de código aberto que foi projetado para armazenar e consultar com eficiência uma grande quantidade de dados com registro de data e hora. Esses dados podem ser dados de sensor de IoT, dados analíticos em tempo real, dados de métricas de aplicações ou dados de monitoramento DevOps. Ele fornece gerenciamento de ciclo de vida de dados automático expirando e excluindo dados antigos. Ele fornece recursos de gravação e consulta usando linguagem de consulta semelhante a SQL, API HTTP e bibliotecas de clientes. Confira os detalhes aqui.

Grafana: O Grafana é um painel métrico de código aberto e editor gráfico. O Grafana também suporta alertas e notificações de métricas. Ele suporta muitas fontes de dados como Graphite, InfluxDB, OpenTSDB, Prometheus, Elasticsearch e CloudWatch. Muitos painéis e plugins (tanto de código aberto quanto comercial) estão disponíveis no site do Grafana. Para mais detalhes sobre o Grafana, visite o site.

Arquitetura do sistema

A aplicação Spark é executada numa rede de clusters, que pode consistir em alguns nós até milhares de nós. Para coletar métricas desse sistema distribuído e enviar as métricas para outros sistemas para análise adicional, precisamos de uma arquitetura com baixa acoplação e tolerante a falhas. Publicar as métricas para um tópico Kafka é uma das melhores soluções. O profiler de JVM da Uber vem com um "KafkaOutputReporter" que pode ser usado para essa finalidade. Outra abordagem de solução para armazenar e ler as métricas é usando o InfluxDB. O InfluxDB fornece a API HTTP, que pode ser usada para consultar e gravar no banco de dados. Essa API suporta autenticação básica e de token JWT e URLs podem ser configuradas para acesso HTTPS. O "InfluxDBOutputReporter" que estamos discutindo neste artigo invoca ponto de escrita HTTP para armazenar as métricas coletadas por diferentes profilers. O Grafana fornece um plug-in de fonte de dados avançado para o InfluxDB. O Grafana usa o ponto de consulta HTTP para buscar dados de métricas do InfluxDB e exibe os dados no painel na forma de gráficos e tabelas. Esses gráficos e tabelas são atualizados automaticamente em intervalos fixos e os intervalos de tempo podem ser configurados nas configurações do Grafana.

O diagrama de arquitetura do sistema de monitoramento de desempenho da aplicação Spark usando o Uber JVM Profiler, InfluxDB e Grafana é ilustrado na Figura 1 abaixo.

Figura 1. Diagrama da arquitetura do sistema de monitoramento de performance

Tecnologias e ferramentas

A tabela a seguir mostra as tecnologias e ferramentas usadas para o sistema de monitoramento de desempenho.

Por favor, consulte a documentação de instalação e configuração das ferramentas.

Projeto e implementação

As seções a seguir fornecem detalhes de projeto e implementação de um sistema de monitoramento de desempenho para aplicações Spark. O profiler de JVM do Uber coleta métricas do guia e dos executores e essas métricas possuem informações detalhadas como papel (driver / executor), Uuid do processo e host. Esses detalhes são úteis para identificar os diferentes sistemas e analisar suas métricas. No InfluxDB podemos usar esses detalhes para consultar as métricas de desempenho de diferentes executores. Podemos adicionar Uuid do processo em tags InfluxDB, aumentando o desempenho da consulta. Primeiro, será criado o banco de dados "métricas" no InfluxDB, depois será adicionado um "InfluxDBOutputReporter" na base de código do profiler de JVM e finalmente será configurado o painel e os gráficos do Grafana.

Criar banco de dados de métricas no InfluxDB

• Inicie o servidor InfluxDB que por padrão é executado na porta 8086. Por exemplo, no sistema Ubuntu, abra um terminal e execute o comando "influxd".

/user/bin$ sudo influxd

• Quando o servidor estiver em funcionamento, inicie o utilitário "influx" em outro terminal. E, em seguida, execute o comando para criar o banco de dados de "métricas".

/user/bin$ sudo influx
CREATE DATABASE metrics

Implementar o InfluxDBOutputReporter

Será implementado o "InfluxDBOutputReporter" na base de código JVM. Consulte o arquivo InfluxDBOutputReporter.java na ramificação "influxdb_reporter" no GitHub para obter os detalhes de implementação discutidos nesta seção.

• Será usada a biblioteca influxdb-java para interagir com o servidor de banco de dados InfluxDB. Ele usa a API HTTP para ler e gravar no banco de dados do InfluxDB. Atualize o arquivo pom.xml e inclua a dependência maven para a biblioteca influxdb-java.
• Defina propriedades de host, porta e banco de dados para o servidor de banco de dados InfluxDB na classe InfluxDBOutputReporter. Os valores padrão para essas propriedades são "127.0.0.1", "8086" e "metrics", respectivamente. Crie conexão com o InfluxDB usando API fornecida pela biblioteca influxdb-java.
• A classe InfluxDBOutputReporter precisa implementar a interface com.uber.profiling.Reporter. Precisa substituir o método public void (String profilerName, Map <String, Object> métricas) e public void close (). O "profilerName" será o nome da medida no banco de dados de "metrics".
• Como o InfluxDB usa o protocolo Line para armazenar pares chave=valor, precisa processar os dados de métricas disponíveis em Map <String, Object> enviados pelo Profiler e convertê-los em um formulário para que ele possa ser armazenado em um Ponto de série única. Podemos usar o valor da propriedade "name" das métricas, se estiverem disponíveis, para formar o nome do campo. Se a propriedade "name" não estiver disponível, podemos usar um valor de contador. O InfluxDB suporta expressões regulares para o nome do campo na consulta, portanto, consultar esses campos não será um problema.
• Crie Point e Batchpoint usando APIs da biblioteca influxdb-java e grave o Batchpoints no banco de dados InfluxDB..
• Para ler as propriedades de conexão do banco de dados do arquivo yaml precisa usar as classes com.uber.profiling.YamlConfigProvider e com.uber.profiling.Argument. Adicione uma referência da classe InfluxDBOutputReporter na classe Argument e chame o método setProperties para passar as propriedades vindas do arquivo yaml. Um arquivo yaml de amostra para influxdb está disponível em recursos / influxdb.yaml no GitHub.

Adicionar fonte de dados e painel no Grafana

Esta seção fornece breves detalhes sobre as etapas necessárias para adicionar um gráfico para dados de métrica do InfluxDB no Grafana. Por favor, verifique os documentos do Grafana para mais detalhes:

• Inicie o servidor Grafana. No Ubuntu, podemos executar o comando abaixo. Por padrão, ele é executado na porta 3000.

sudo service grafana-server start

• Abra o endereço http://localhost: 3000/ em um navegador e crie uma fonte de dados para o InfluxDB. Adicione o nome como "InfluxDBDataSource", tipo como "InfluxDB", a URL padrão para o InfluxDB é "http://localhost:8086" e o nome do banco de dados será "metrics".
• Crie um novo painel clicando em Graph, clique em Edit e adicione a consulta para o gráfico. Abaixo está uma consulta de amostra para o banco de dados InfluxDB.

select "heapMemoryCommitted" as Committed, "heapMemoryTotalUsed" as Used from "metrics"."autogen"."CpuAndMemory" where "role" = 'driver' AND time > now() - 5m

• O Grafana fornece a opção de definir variáveis de modelo que podem ser passadas na consulta. Isso será útil para exibir dados de vários executores no painel. Por exemplo, podemos criar variáveis para "executorProcessUuid" e "timeInterval" e usá-las na consulta conforme abaixo.

select "heapMemoryCommitted" as Committed, "heapMemoryTotalUsed" as Used from "metrics"."autogen"."CpuAndMemory” where "processUuid" =~ /^$executorProcessUuid$/ AND time > now() - $timeInterval

Um arquivo JSON de amostra do painel Spark-Metrics foi disponibilizado no GitHub. Este arquivo pode ser importado no servidor Grafana. Abra a URL http://localhost:3000/dashboard/import em um navegador e clique em "Upload.json File".

Criação e implementação

Esta seção descreve as etapas para criar e implantar um sistema de monitoramento de desempenho. O código da aplicação pode ser clonado a partir da ramificação "influxdb_reporter" no GitHub.

• Use o comando abaixo para construir o profiler da JVM com "InfluxDBOutputReporter".

mvn clean package

• Copie o arquivo JVM Profiler-0.0.9.jar criado pelo maven build para algum diretório. (por exemplo, /opt/profiler). Pode manter o arquivo influxdb.yaml também neste diretório.
• Será usado a aplicação JavaNetworkWordCount existente para criação de perfil, fornecido com o Apache Spark. O código-fonte está disponível em /spark-2.3.1-bin-hadoop2.7/examples/src/main/java/org/apache/spark/examples/streaming.
• Para executar a aplicação JavaNetworkWordCount precisamos executar o servidor Netcat usando o comando abaixo.

nc -lk 9999

• Ir para /spark-2.3.1-bin-hadoop2.7/sbin e iniciar o mestre usando o comando abaixo.

./start-master.sh

• Pode-se obter a URL do mestre no arquivo de log. Passe esta URL para o comando para iniciar o executor.

./start-slave.sh -c 2 spark://192.168.1.6:7077

• Ir para /spark-2.3.1-bin-hadoop2.7/bin e executar o comando abaixo. Esse comando executará o aplicativo JavaNetworkWordCount e iniciará os profilers da JVM para o driver e o executor. Verifique a página de leia-me do GitHub do profiler de JVM do Uber para obter detalhes sobre os parâmetros.

spark-submit --master spark://192.168.1.6:7077 --conf "spark.driver.extraJavaOptions=-javaagent:/opt/profiler/jvm-profiler-0.0.9.jar=reporter=com.uber.profiling.reporters.InfluxDBOutputReporter,metricInterval=5000,sampleInterval=5000,ioProfiling=true" --conf "spark.executor.extraJavaOptions=-javaagent:/opt/profiler/jvm-profiler-0.0.9.jar=reporter=com.uber.profiling.reporters.InfluxDBOutputReporter,tag=influxdb,configProvider=com.uber.profiling.YamlConfigProvider,configFile=/opt/profiler/influxdb.yaml,metricInterval=5000,sampleInterval=5000,ioProfiling=true" --class org.apache.spark.examples.streaming.JavaSqlNetworkWordCount ../examples/jars/spark-examples_2.11-2.3.1.jar localhost 9999

• Opcionalmente, pode-se executar a aplicação com o arquivo yaml. Passe os parâmetros "configProvider" e "configFile" no comando como abaixo.

spark-submit --master spark://192.168.1.6:7077 --conf "spark.driver.extraJavaOptions=-javaagent:/opt/profiler/jvm-profiler-0.0.9.jar=reporter=com.uber.profiling.reporters.InfluxDBOutputReporter,configProvider=com.uber.profiling.YamlConfigProvider,configFile=/opt/profiler/influxdb.yaml,metricInterval=5000,sampleInterval=5000,ioProfiling=true" --conf "spark.executor.extraJavaOptions=-javaagent:/opt/profiler/jvm-profiler-0.0.9.jar=reporter=com.uber.profiling.reporters.InfluxDBOutputReporter,tag=influxdb,configProvider=com.uber.profiling.YamlConfigProvider,configFile=/opt/profiler/influxdb.yaml,metricInterval=5000,sampleInterval=5000,ioProfiling=true" --class org.apache.spark.examples.streaming.JavaSqlNetworkWordCount ../examples/jars/spark-examples_2.11-2.3.1.jar localhost 9999

• Vá para o para o terminal "influx" e execute os comandos abaixo.

use metrics
show measurements

• Será recebido o nome de "Métricas" criadas no banco de dados de "metrics", conforme abaixo.

CpuAndMemory
IO
ProcessInfo
Stacktrace

Pode-se verificar os dados preenchidos nestas Métricas. Por exemplo, usar o comando abaixo para buscar um único registro da medição CPU e Memória.

select * from CpuAndMemory limit 1

• Vá para o painel Grafana para Spark. Os gráficos para métricas devem estar preenchidos com dados. Abaixo está o painel de amostras.

Figura 2: Amostra do painel Grafana de métricas Spark

Prós e contras

O sistema de monitoramento de desempenho discutido neste artigo coleta métricas de sistemas e aplicações diferentes usando o profile de JVM do Uber e as armazena no banco de dados de séries temporais InfluxDB. O banco de dados de séries temporais fornece políticas de retenção de dados, consultas contínuas, agregações de tempo flexíveis e manuseio eficiente de milhões de registros para processamento em tempo real e em lote. Esses dados de séries temporais nos ajudam a analisar como o sistema mudou no passado, como o sistema está se comportando no presente e ajudam a prever como pode mudar no futuro. Podemos identificar os padrões de falhas correlacionando essas métricas coletadas ao longo do tempo. Foi criado um painel usando o Grafana que facilita o acesso e a navegação em diferentes tipos de métricas. A equipe de DevOps pode usar estes gráficos para correlacionar as diferentes métricas para entender o comportamento do sistema e identificar pontos com alto índice de leitura nos dados. Isso ajuda na manutenção da conformidade e na obtenção dos contratos de nível de serviço de uma aplicação. Em geral, esse sistema de monitoramento de desempenho ajuda a alcançar o monitoramento contínuo do sistema.

As limitações deste sistema de monitoramento de desempenho estão alinhadas com as limitações que temos para profilers baseados em agente. O profiler baseado em agente pode consumir uma certa quantidade de recursos de computação. Às vezes pode ser necessário solucionar problemas e aplicar correções, o que pode ser difícil para grandes sistemas distribuídos geograficamente. Pode ser necessário concluir este trabalho de casa antes de instalar os agentes no sistema de produção. Também deve-se considerar a segurança, escalabilidade e disponibilidade do sistema de monitoramento de desempenho, ou seja, monitorar o sistema de monitoramento. A maioria desses desafios pode ser superada pelo projeto com ajustes adequados da aplicação e do sistema. Se não tiver permissão para instalar agentes do profilers em um sistema de produção, considere um sistema sem agente, mas ele tem suas próprias limitações, como métricas menos granulares e maior sobrecarga de rede.

Conclusão

Para aplicações Spark complexas, identificação, depuração e resolução dos problemas de produção não são tarefas fáceis e precisamos de um sistema de monitoramento de desempenho eficaz que possa nos ajudar com isso. O profiler de JVM do Uber é um ótimo complemento para a comunidade de código aberto. Esse profiler de JVM pode ser estendido e um novo reporter para publicação de métricas pode ser incluído. As métricas de desempenho para aplicações Spark coletadas por diferentes profilers podem ser armazenadas no InfluxDB. O "InfluxDBOutputReporter" discutido neste artigo usa a API HTTP para gravar métricas para os guias e executores do Spark para o InfluxDB. O Grafana fornece um plug-in para o InfluxDB e usa a API HTTP para consultar as métricas. Pode-se criar um painel com gráficos e tabelas para essas métricas, que podem ser atualizadas automaticamente em intervalos fixos. O código para "InfluxDBOutputReporter" está disponível no Github do projeto assimcomo o arquivo Spark-InfluxDB-Grafana.json também está disponível aqui.

Referências

• Profiler de JVM do Uber - Blog
• InfluxDB -Documentação
• Grafana -Documentação

Sobre o autor

Amit Baghel é arquiteto de software com mais de 17 anos de experiência em projeto e desenvolvimento de aplicações e produtos corporativos em torno do ecossistema Java. Seu foco atual é em IoT, Cloud Computing, Soluções de Big Data, Microserviços, DevOps e Integração e Entrega Contínua. Baghel pode ser localizado via e-mail.