Olá, Habr! Hoje vamos construir um sistema que processará fluxos de mensagens do Apache Kafka usando o Spark Streaming e gravará o resultado na base de dados em nuvem da AWS RDS.
Vamos imaginar que uma instituição de crédito nos encarrega de processar transações de entrada 'em tempo real' em todas as suas filiais. Isso pode ser feito com o objetivo de calcular rapidamente a posição de moedas abertas para o tesouro, limites ou resultados financeiros de transações, etc.
Como implementar esse caso sem o uso de magia e feitiços — leia abaixo! Vamos lá!
Introdução
Sem dúvida, o processamento de grandes volumes de dados em tempo real oferece amplas oportunidades de aplicação em sistemas modernos. Uma das combinações mais populares para isso é a parceria entre Apache Kafka e Spark Streaming, onde o Kafka cria um fluxo de pacotes de mensagens de entrada e o Spark Streaming processa esses pacotes em intervalos de tempo definidos.
Para aumentar a resiliência da aplicação, vamos usar pontos de verificação — checkpoints. Com esse mecanismo, quando o módulo Spark Streaming precisar recuperar dados perdidos, ele só precisará voltar ao último ponto de verificação e retomar os cálculos a partir daí.
Arquitetura do sistema em desenvolvimento
Componentes utilizados:
- — é um sistema distribuído de troca de mensagens com publicação e assinatura. Serve tanto para o consumo autônomo quanto online das mensagens. Para evitar a perda de dados, as mensagens do Kafka são armazenadas em disco e replicadas dentro do cluster. O sistema Kafka é construído sobre o serviço de sincronização ZooKeeper;
- — componente Spark para processamento de dados em tempo real. O módulo Spark Streaming é construído com uma arquitetura de "micro-batch", onde um fluxo de dados é interpretado como uma sequência contínua de pequenos pacotes de dados. O Spark Streaming aceita dados de diferentes fontes e os combina em pequenos pacotes. Novos pacotes são criados em intervalos regulares de tempo. No início de cada intervalo, um novo pacote é criado, e todos os dados recebidos durante esse intervalo são incluídos no pacote. Ao final do intervalo, o aumento do pacote é interrompido. O tamanho do intervalo é determinado por um parâmetro chamado intervalo de lote (batch interval);
- — combina o processamento relacional com a programação funcional do Spark. Dados estruturados são aquilo que possui um esquema, ou seja, um conjunto único de campos para todos os registros. O Spark SQL suporta a entrada de várias fontes de dados estruturados e, graças à informação sobre o esquema, ele pode extrair eficientemente apenas os campos necessários dos registros, além de fornecer interfaces de API DataFrame;
- — é um banco de dados relacional em nuvem relativamente barato, um serviço web que facilita a configuração, operação e escalabilidade, administrado diretamente pela Amazon.
Instalação e execução do servidor Kafka
Antes de usar o Kafka, é necessário certificar-se de que o Java está instalado, pois o funcionamento é realizado por meio da JVM:
sudo apt-get update
sudo apt-get install default-jre
java -version
Vamos criar um novo usuário para trabalhar com o Kafka:
sudo useradd kafka -m
sudo passwd kafka
sudo adduser kafka sudo
Em seguida, baixamos a distribuição do site oficial do Apache Kafka:
wget -P /YOUR_PATH "http://apache-mirror.rbc.ru/pub/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz"Descompactamos o arquivo baixado:
tar -xvzf /YOUR_PATH/kafka_2.12-2.2.0.tgz
ln -s /YOUR_PATH/kafka_2.12-2.2.0 kafka
O próximo passo é opcional. O fato é que as configurações padrão não permitem aproveitar totalmente todos os recursos do Apache Kafka. Por exemplo, excluir tópicos, categorias, grupos que podem ter mensagens publicadas. Para mudar isso, vamos editar o arquivo de configuração:
vim ~/kafka/config/server.propertiesAdicione o seguinte ao final do arquivo:
delete.topic.enable = trueAntes de iniciar o servidor Kafka, é necessário iniciar o servidor ZooKeeper; usaremos um script auxiliar que vem junto com a distribuição do Kafka:
Cd ~/kafka
bin/zookeeper-server-start.sh config/zookeeper.properties
Depois que o ZooKeeper iniciar com sucesso, em um terminal separado inicie o servidor Kafka:
bin/kafka-server-start.sh config/server.propertiesVamos criar um novo tópico chamado Transaction:
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic transactionVamos garantir que o tópico com o número necessário de partições e replicação foi criado:
bin/kafka-topics.sh --describe --zookeeper localhost:2181 
Vamos pular os momentos de teste do produtor e consumidor para o novo tópico. Mais detalhes sobre como testar o envio e recebimento de mensagens estão escritos na documentação oficial — . E agora passamos para a escrita do produtor em Python usando a API KafkaProducer.
Escrevendo o produtor
O produtor irá gerar dados aleatórios — 100 mensagens a cada segundo. Dados aleatórios são entendidos como um dicionário composto por três campos:
- Filial — nome do ponto de venda da instituição financeira;
- Currency — moeda da transação;
- Amount — montante da transação. O montante será um número positivo se for a compra de moeda pelo Banco, e negativo se for a venda.
O código para o produtor é o seguinte:
from numpy.random import choice, randint
def get_random_value():
new_dict = {}
branch_list = ['Kazan', 'SPB', 'Novosibirsk', 'Surgut']
currency_list = ['RUB', 'USD', 'EUR', 'GBP']
new_dict['branch'] = choice(branch_list)
new_dict['currency'] = choice(currency_list)
new_dict['amount'] = randint(-100, 100)
return new_dict
Em seguida, usando o método send, enviamos a mensagem ao servidor, para o tópico que precisamos, no formato JSON:
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
value_serializer=lambda x:dumps(x).encode('utf-8'),
compression_type='gzip')
my_topic = 'transaction'
data = get_random_value()
try:
future = producer.send(topic=my_topic, value=data)
record_metadata = future.get(timeout=10)
print('--> A mensagem foi enviada para um tópico:
{}, partição: {}, offset: {}'
.format(record_metadata.topic,
record_metadata.partition,
record_metadata.offset))
except Exception as e:
print('--> Parece que ocorreu um erro: {}'.format(e))
finally:
producer.flush()
Ao executar o script, obtemos as seguintes mensagens no terminal:
Isso significa que tudo está funcionando como desejávamos — o produtor gera e envia mensagens para o tópico que precisamos.
O próximo passo será instalar o Spark e processar esse fluxo de mensagens.
Instalação do Apache Spark
Apache Spark — é uma plataforma de computação em cluster universal e de alto desempenho.
Em termos de desempenho, o Spark supera implementações populares do modelo MapReduce, enquanto oferece suporte a uma gama mais ampla de tipos de computação, incluindo consultas interativas e processamento em fluxo. A velocidade desempenha um papel crucial no processamento de grandes volumes de dados, pois é a velocidade que permite trabalhar de forma interativa, sem gastar minutos ou horas esperando. Uma das principais vantagens do Spark, que garante essa alta velocidade, é a capacidade de realizar cálculos na memória.
Este framework é escrito em Scala, portanto, é necessário instalá-la primeiro:
sudo apt-get install scalaBaixamos o pacote do Spark do site oficial:
wget "http://mirror.linux-ia64.org/apache/spark/spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz"Descompactamos o arquivo:
sudo tar xvf spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz -C /usr/local/sparkAdicionar o caminho para o Spark no arquivo bash:
vim ~/.bashrcAdicione as seguintes linhas no editor:
SPARK_HOME=/usr/local/spark
export PATH=$SPARK_HOME/bin:$PATH
Execute o comando abaixo após fazer as alterações no bashrc:
source ~/.bashrcImplantação do PostgreSQL na AWS
Agora precisamos implantar o banco de dados onde vamos carregar as informações processadas dos fluxos. Para isso, usaremos o serviço AWS RDS.
Acesse o console da AWS —> AWS RDS —> Bancos de Dados —> Criar banco de dados:
Escolhemos PostgreSQL e clicamos no botão Avançar:
Como este exemplo é discutido exclusivamente para fins educacionais, usaremos um servidor gratuito "mínimo" (Free Tier):
Em seguida, marcamos a opção no bloco Free Tier, e, após isso, nos será automaticamente oferecido uma instância da classe t2.micro — embora seja bastante limitada, é gratuita e atende perfeitamente à nossa necessidade:
A seguir, estão informações muito importantes: o nome da instância do banco de dados, o nome do usuário master e sua senha. Nomearemos a instância: myHabrTest, usuário master: habr, senha: habr12345 e clicamos no botão Avançar:
Na próxima página, estão os parâmetros que controlam a acessibilidade do nosso servidor de banco de dados externamente (Acessibilidade Pública) e a disponibilidade das portas:
Vamos criar uma nova configuração para o grupo de segurança VPC que permitirá o acesso ao nosso servidor de banco de dados pela porta 5432 (PostgreSQL).
Vamos para uma nova janela do navegador para o console da AWS na seção VPC Dashboard —> Grupos de Segurança —> Criar grupo de segurança:
Definimos o nome para o grupo de segurança — PostgreSQL, a descrição, indicamos a qual VPC este grupo deve estar associado e clicamos no botão Criar:
Preenchemos para o grupo recém-criado as regras de entrada para a porta 5432, conforme mostrado na imagem abaixo. Você pode não especificar a porta manualmente, mas escolher PostgreSQL na lista suspensa Tipo.
Estritamente falando, o valor ::/0 significa a disponibilidade do tráfego de entrada para o servidor de todo o mundo, o que não é exatamente correto, mas para o propósito deste exemplo, vamos permitir-nos usar essa abordagem:
Voltamos à página do navegador, onde temos aberta a opção 'Configurar configurações avançadas' e escolhemos na seção grupos de segurança VPC —> Escolher grupos de segurança VPC existentes —> PostgreSQL:
Em seguida, na seção Opções do banco de dados —> Nome do banco de dados —> definimos o nome — habrDB.
Os outros parâmetros, exceto talvez a desativação das cópias de segurança (período de retenção de backup — 0 dias), monitoramento e Performance Insights, podem ser deixados como padrão. Clicamos no botão Criar banco de dados:
Processador de fluxos
A etapa final será o desenvolvimento de uma tarefa Spark que processará novos dados provenientes do Kafka a cada dois segundos e registrará o resultado no banco de dados.
Como mencionado acima, os pontos de verificação (checkpoints) são o principal mecanismo no Spark Streaming que deve ser configurado para garantir a resiliência. Usaremos checkpoints e, em caso de falha do processo, a módulo Spark Streaming para recuperar os dados perdidos precisará apenas retornar ao último ponto de verificação e retomar os cálculos a partir dele.
Um ponto de verificação pode ser ativado definindo um diretório em um sistema de arquivos resiliente e confiável (por exemplo, HDFS, S3, etc.), onde as informações do ponto de verificação serão salvas. Isso é feito com, por exemplo:
streamingContext.checkpoint(checkpointDirectory)No nosso exemplo, usaremos a seguinte abordagem, ou seja, se o checkpointDirectory existir, o contexto será recriado a partir dos dados do ponto de verificação. Se o diretório não existir (ou seja, estiver sendo executado pela primeira vez), a função functionToCreateContext será chamada para criar um novo contexto e configurar DStreams:
from pyspark.streaming import StreamingContext
context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)
Criamos um objeto DirectStream para se conectar ao tópico 'transaction' usando o método createDirectStream da biblioteca KafkaUtils:
from pyspark.streaming.kafka import KafkaUtils
sc = SparkContext(conf=conf)
ssc = StreamingContext(sc, 2)
broker_list = 'localhost:9092'
topic = 'transaction'
directKafkaStream = KafkaUtils.createDirectStream(ssc,
[topic],
{"metadata.broker.list": broker_list})
Analisando os dados de entrada no formato JSON:
rowRdd = rdd.map(lambda w: Row(branch=w['branch'],
currency=w['currency'],
amount=w['amount']))
testDataFrame = spark.createDataFrame(rowRdd)
testDataFrame.createOrReplaceTempView("treasury_stream")
Usando o Spark SQL, fazemos uma simples agrupamento e mostramos o resultado no console:
select
from_unixtime(unix_timestamp()) as curr_time,
t.branch as branch_name,
t.currency as currency_code,
sum(amount) as batch_value
from treasury_stream t
group by
t.branch,
t.currency
Obtendo o texto da consulta e executando-a através do Spark SQL:
sql_query = get_sql_query()
testResultDataFrame = spark.sql(sql_query)
testResultDataFrame.show(n=5)
Em seguida, salvamos os dados agregados em uma tabela no AWS RDS. Para salvar os resultados da agregação na tabela do banco de dados, usaremos o método write do objeto DataFrame:
testResultDataFrame.write
.format("jdbc")
.mode("append")
.option("driver", 'org.postgresql.Driver')
.option("url","jdbc:postgresql://myhabrtest.ciny8bykwxeg.us-east-1.rds.amazonaws.com:5432/habrDB")
.option("dbtable", "transaction_flow")
.option("user", "habr")
.option("password", "habr12345")
.save()
Algumas palavras sobre a configuração da conexão com o AWS RDS. O usuário e a senha foram criados na etapa "Implantação do AWS PostgreSQL". Como URL do servidor de banco de dados, devemos usar o Endpoint que é exibido na seção Connectivity & security:
Para uma integração correta entre Spark e Kafka, a job deve ser executada através do spark-submit utilizando o artefato spark-streaming-kafka-0-8_2.11. Adicionalmente, também aplicaremos o artefato para interagir com o banco de dados PostgreSQL, que serão passados através de —packages.
Para dar flexibilidade ao script, vamos também definir como parâmetros de entrada o nome do servidor de mensagens e o tópico do qual queremos receber dados.
Assim, chegou a hora de iniciar e verificar a funcionalidade do sistema:
spark-submit
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2,
org.postgresql:postgresql:9.4.1207
spark_job.py localhost:9092 transaction
Tudo funcionou! Como pode ser visto na imagem abaixo — durante a execução do aplicativo, novos resultados de agregação são exibidos a cada 2 segundos, porque configuramos o intervalo de lotes para 2 segundos ao criar o objeto StreamingContext:
Em seguida, fazemos uma simples consulta ao banco de dados para verificar a presença de registros na tabela transaction_flow:
Conclusão
Neste artigo, foi apresentado um exemplo de processamento de informações em tempo real usando Spark Streaming em conjunto com Apache Kafka e PostgreSQL. Com o aumento do volume de dados de diversas fontes, é difícil subestimar o valor prático do Spark Streaming para a criação de aplicações em tempo real.
Você pode encontrar o código-fonte completo no meu repositório em .
Estou ansioso para discutir este artigo, aguardo seus comentários e espero críticas construtivas de todos os leitores interessados.
Desejo sucesso!
Ps. Inicialmente, planejava usar um banco de dados PostgreSQL local, mas considerando meu carinho pela AWS, decidi migrar o banco de dados para a nuvem. No próximo artigo sobre este tema, mostrarei como implementar todo o sistema descrito acima na AWS usando AWS Kinesis e AWS EMR. Fiquem atentos às novidades!
Fonte: habr.com
