Apache Kafka — Aprendendo na prática

Photo by David Wirzba on Unsplash

Recentemente fiz um post sobre Apache Kafka no qual falei os principais conceitos(Producer, Consumer, Topic,etc) e também citei alguns dos possíveis casos de uso. A ideia deste post é dar continuidade nestes estudos, aqui vamos ver na prática o funcionamento desta ferramenta criada pelo LinkedIn.

Preparando ambiente

Para iniciar nesta jornada, o primeiro passo será instalar e configurar o Kafka. Quando eu estava estudando sobre, eu achei duas possibilidades de fazer isso:

1. Instalando como serviço na sua máquina local, no site oficial do Kafka mostra como: https://kafka.apache.org/quickstart
2. Utilizando um dos tanto containers disponibilizados pela comunidade. A Confluent( empresa fundada pelo criadores do Kafka e que oferece serviços de Kafka-as-a-service) disponibilizou um tutorial mostrando como utilizar o container mantido por eles: https://docs.confluent.io/current/installation/docker/docs/quickstart.html#getting-started-with-docker-compose

Pela comodidade e facilidade do uso de containers — já falei disso neste post e também neste outro — eu optei pela opção dois. O restante dessa sessão será uma reprodução do tutorial disponibilizado pela Confluent.

Pré-requisitos

Antes de começar, é necessário instalar as ferramentas que serão utilizadas, que é o Docker, o Docker Compose e o Git.

Dica: No caso do Docker e Docker Compose, não instale através de instaladores de pacotes ( como por exemplo o apt get). Pois há chances da versão do pacote estar desatualizada ( isso já aconteceu comigo).

Abaixo segue o link das instalações oficiais:

• Docker CE(Community Edition)
• Docker Compose
• Git

Instalando e iniciando o Kafka

Para iniciar o Kafka utilizaremos um arquivodocker-compose criado e mantido pela Confluent, disponível no GitHub deles.

O primeiro passo é clonar o repositório:

git clone https://github.com/confluentinc/cp-docker-images

Após clonado, navegue até a pasta cp-docker-images/examples/kafka-single-node. Esta pasta conterá o seguinte arquivo docker-compose.yml:

Não vou entrar nos detalhes de implementação deste arquivo, mas gostaria de chamar atenção para o que ele faz: Ele basicamente irá inicializar dois serviços, o Zookeeper e o Kafka.

Mas afinal, o que é o Zookeeper? Resumidamente, o Apache Zookeeper é um serviço centralizado para manter informações de configurações e nomenclaturas entre serviços distribuídos. O Kafka utiliza o Zookeeper para sincronizar as configurações entre diferentes clusters.

Agora dentro da pasta, execute o comando docker-compose up -d. O -d serve para executar em segundo plano. Agora vamos executar o comando docker-compose ps para verificar se os serviços estão executando corretamente. Deverá aparecer algo como abaixo:

Name                         Command            State   Ports
----------------------------------------------------------------
kafka-single-node_kafka_1       /etc/confluent/docker/run   Up
kafka-single-node_zookeeper_1   /etc/confluent/docker/run   Up

Olhando os logs, é possível verificar se os serviços estão funcionando conforme o esperado. O comando abaixo serve para analisar os logs do serviço Zookeeper.

$ docker-compose logs zookeeper | grep -i binding

Deverá aparecer algo assim:

zookeeper_1  | [2018-06-26 01:06:59,447] INFO binding to port 0.0.0.0/0.0.0.0:32181 (org.apache.zookeeper.server.NIOServerCnxnFactory)

Vamos também analisar a saúde do serviço do Kafka.

docker-compose logs kafka | grep -i started

Deverá aparecer algo como abaixo:

kafka_1      | [2018-06-27 20:03:50,641] INFO [SocketServer brokerId=1] Started 1 acceptor threads (kafka.network.SocketServer)
kafka_1      | [2018-06-27 20:03:50,898] INFO [SocketServer brokerId=1] Started processors for 1 acceptors (kafka.network.SocketServer)
kafka_1      | [2018-06-27 20:03:50,900] INFO [KafkaServer id=1] started (kafka.server.KafkaServer)
kafka_1      | [2018-06-27 20:03:50,911] INFO [ReplicaStateMachine controllerId=1] Started replica state machine with initial state -> Map() (kafka.controller.ReplicaStateMachine)
kafka_1      | [2018-06-27 20:03:50,914] INFO [PartitionStateMachine controllerId=1] Started partition state machine with initial state -> Map() (kafka.controller.PartitionStateMachine)

Fazendo um test drive

Agora que já preparamos o ambiente e estamos com tudo rodando, vamos testar e ver ele funcionando via comando de linha.

Criando um Topic

Para verificar se o nosso cluster/broker estão funcionando corretamente, vamos criar um Topic. Para fazer isto vamos executar um comando de linha diretamente no container que está rodando o Kafka(utilizando o docker-compose exec). O comando abaixo cria um Topic chamado “meu-topico-legal”.

docker-compose exec kafka  \
kafka-topics --create --topic meu-topico-legal --partitions 1 --replication-factor 1 --if-not-exists --zookeeper localhost:32181

Se quiser confirmar se o Topic foi criado, execute o comando abaixo:

docker-compose exec kafka  \
  kafka-topics --describe --topic meu-topico-legal --zookeeper localhost:32181

Deverá aparecer algo assim:

Topic:meu-topico-legal PartitionCount:1 ReplicationFactor:1 Configs:
 Topic: meu-topico-legal Partition: 0 Leader: 1 Replicas: 1 Isr: 1

Note que a porta do Zookeeper é a 32181. Isto está configurado no arquivo docker-compose.yml .

Produzindo mensagens com o Producer

Agora que o Topic já está criado, vamos enviar mensagens para ele. O comando abaixo irá enviar 100 mensagens para o Topic.

docker-compose exec kafka  \
  bash -c "seq 100 | kafka-console-producer --request-required-acks 1 --broker-list localhost:29092 --topic meu-topico-legal && echo 'Produced 100 messages.'"

Se tudo funcionar corretamente, você deverá ver uma mensagem como a abaixo:

Produced 100 messages.

Note que a porta do broker é a 29092. Isto está configurado no arquivo docker-compose.yml .

Consumindo mensagens com o Consumer

Agora, para completar o circulo, vamos ler novamente estas mensagens, rode o comando abaixo:

docker-compose exec kafka  \
  kafka-console-consumer --bootstrap-server localhost:29092 --topic meu-topico-legal --from-beginning --max-messages 100

Deverá aparecer algo assim:

1 
....
....
100
Processed a total of 100 messages

Com isso concluímos o test-drive inicial de Kafka, agora vamos ver o Kafka na prática, utilizando as bibliotecas para produzir e consumir as mensagens.

#PartiuProgramar

A Figura 1 demonstra como será a estrutura do nosso exemplo. Em Python, desenvolveremos a aplicação Producer. Ela será um loop infinito que produzirá números randômicos( entre 1 e 999). No lado dos Consumers, desenvolveremos duas aplicações utilizando Node.js. O primeiro dos Consumers fará um somatório de todas as mensagens recebidas. O segundo, fará a média das mensagens.

Figura 1 — Ilustração do cenário

Estrutura do tutorial

Por questões de praticidade, vamos criar somente uma pasta tutorial e dentro dela haverá duas subpastas, producer e consumer.A estrutura deverá ser assim:

|-- tutorial
    |-- producer
        |-- producer.py
    |-- consumer
        |-- consumer.js

Producer (Python)

Antes de iniciar, verifique se você possui instalado os seguintes pré-requisitos:

• python
• pip
• virtualenv

Estando dentro da pasta tutorial, crie o virtualenv e o habilite logo em seguida:

virtualenv -p python3 .env3
source .env3/bin/activate

Agora instale o pacote kafka-python ( client Kafka para python que utilizaremos neste tutorial):

pip install kafka-python

Abaixo o código do producer/producer.py:

Vamos analisar alguns trechos de código:

O KafkaProducer é inicializado com dois parâmetros:

• bootstrap-servers : A lista dos brokers que serão enviadas as mensagens. No nosso exemplo é somente um broker( que está configurado no docker-compose.yml).
• value_serializer : O método de serialization das mensagens. Para simplificar, no nosso exemplo estamos transportando elas como string. Porém é comum serializar as mensagens usando JSON ou Protobuf.

O método producer.send() é utilizado para enviar as mensagens, o primeiro parâmetro é o tópico e o segundo é a mensagem. Note que o tópico ainda não existe, no nosso caso, a configuração do Broker está para “auto-criar” caso não exista. Mas isso é configurável. Normalmente os tópicos serão criados separadamente.

Se quiser criar o tópico manualmente, utilize o comando abaixo:

docker-compose exec kafka  \
kafka-topics --create --topic kafka-python-topic --partitions 1 --replication-factor 1 --if-not-exists --zookeeper localhost:32181

Consumer (Node.js)

Novamente vamos começar verificando os pré-requisitos:

• Node.js
• Npm

Após tudo instalado, navegue até a pasta raiz(tutorial) e rode o comando abaixo:

npm install no-kafka

Isto irá instalar o no-kafka, o client Kafka para Node.js.

Abaixo vamos analisar o código de ambos consumers:

consumer/consumer_sum.js

consumer/consumer_avg.js

Ambos são bem semelhantes, vamos analisar a inicialização do modulo Kafka.SimpleConsumer:

• connectionString: Semelhante ao client utilizado para Python, aqui é possível passar um lista de brokers no qual as mensagens serão consumidas. No nosso caso, é somente um broker.
• consumer.subscribe(...): Neste método estamos assinando o tópico, de modo que todas as mensagens serão processadas pelo consumer. No nosso exemplo, para cada novo conjunto de mensagens será executado a function data. Por padrão, o comportamento do consumer será processar somente as novas mensagens, ou seja, aquelas que foram produzidas após o mesmo ser iniciado. Porém é possível configurar para as mensagens do passado serem consumidas também. O trecho abaixo ilustra este cenário:

return consumer.init().then(function () {
      return consumer.subscribe('kafka-python-topic', 0, 
                           {time: Kafka.EARLIEST_OFFSET}, data);
});

Vamos ver a magica acontecer

Agora que está tudo configurado, vamos rodar e ver as mensagens sendo produzidas e processadas.

Primeiro vamos verificar se o kafka e o zookeeper estão rodando corretamente. Para isto, navegue até a pasta cp-docker-images/examples/kafka-single-node e execute o comando abaixo:

docker-compose ps

O resultado deverá aparecer state como up:

Name                         Command            State   Ports
--------------------------------------------------------------------
kafka-single-node_kafka_1       /etc/confluent/docker/run   Up           
kafka-single-node_zookeeper_1   /etc/confluent/docker/run   Up

Para ficar descolado e maneiro, vamos trabalhar com quatro terminais:

1. Um visualizando os logs do Kafka. Para isso, ainda na pasta kafka-single-node rode o seguinte comando: docker-compose logs --f
2. Outro rodando o Producer. Navegue até a pasta raiz do nosso tutorial, habilite o seu virtualenv e execute o comando: python producer/producer.py
3. Outro rodando o Consumer de média. Navegue até a pasta raiz do nosso tutorial e execute o comando consumer/consumer_avg.js
4. Outro rodando o Consumer de somatório. Navegue até a pasta raiz do nosso tutorial e execute o comando consumer/consumer_sum.js

Figura 2 — Kafka, producers e consumers rodando.

Conclusão

A ideia deste tutorial é criar uma base simples para entender os conceitos do Kafka. Espero que a partir daqui fique facil para evoluir os estudos dessa tecnologia. Se você quiser acessar o código completo, ele está disponível no meu github.

Se quiser trocar uma ideia ou entrar em contato comigo, pode me achar no Twitter(@e_ferreirasouza) ou Linkedin.

Grande abraço e até a próxima!