Como ingerir dados no Elasticsearch através do Kafka

Neste artigo, mostramos como integrar o Apache Kafka com o Elasticsearch para ingestão e indexação de dados. Iremos apresentar uma visão geral do Kafka, seu conceito de produtores e consumidores, e criaremos um índice de logs onde as mensagens serão recebidas e indexadas através do Apache Kafka. O projeto foi implementado em Python e o código está disponível no GitHub.

Pré-requisitos

• Docker e Docker Compose: Certifique-se de ter o Docker e o Docker Compose instalados em sua máquina.

• Python 3.x: Para executar os scripts do produtor e do consumidor.

Introdução ao Apache Kafka

O Apache Kafka é uma plataforma de streaming distribuída que permite alta escalabilidade e disponibilidade, além de tolerância a falhas. No Kafka, o gerenciamento de dados ocorre por meio dos seguintes componentes principais:

• Intermediário (Broker): responsável por armazenar e distribuir mensagens entre produtores e consumidores.

• Zookeeper: gerencia e coordena os brokers do Kafka, controlando o estado do cluster, os líderes de partição e as informações do consumidor.

• Tópicos: canais onde os dados são publicados e armazenados para consumo.

• Consumidores e Produtores: enquanto os produtores enviam dados para os tópicos, os consumidores recuperam esses dados.

Esses componentes trabalham juntos para formar o ecossistema Kafka, fornecendo uma estrutura robusta para streaming de dados.

Estrutura do projeto

Para entender o processo de ingestão de dados, dividimos o processo em etapas:

• Provisionamento de infraestrutura: configuração do ambiente Docker para suportar Kafka, Elasticsearch e Kibana.

• Criação do produtor: implementação do produtor Kafka, que envia dados para o tópico de logs.

• Criação do consumidor: desenvolvimento do consumidor Kafka para ler e indexar mensagens no Elasticsearch.

• Validação de ingestão: verificação e validação dos dados enviados e consumidos.

Configuração de infraestrutura com Docker Compose

Utilizamos o Docker Compose para configurar e gerenciar os serviços necessários. A seguir, você encontrará o código Docker Compose que configura cada serviço necessário para a integração do Apache Kafka, Elasticsearch e Kibana, garantindo um processo de ingestão de dados.

version: "3"

services:

  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:latest
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "9094:9094"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST:${HOST_IP}:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.15.1
    container_name: elasticsearch-8.15.1
    environment:
      - node.name=elasticsearch
      - xpack.security.enabled=false
      - discovery.type=single-node
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    volumes:
      - ./elasticsearch:/usr/share/elasticsearch/data
    ports:
      - 9200:9200

  kibana:
    image: docker.elastic.co/kibana/kibana:8.15.1
    container_name: kibana-8.15.1
    ports:
      - 5601:5601
    environment:
      ELASTICSEARCH_URL: http://elasticsearch:9200
      ELASTICSEARCH_HOSTS: '["http://elasticsearch:9200"]'

Você pode acessar o arquivo diretamente do repositório GitHub do Elasticsearch Labs.

Envio de dados com o Kafka Producer

O produtor é responsável por enviar mensagens para o tópico de logs. Ao enviar mensagens em lotes, aumenta-se a eficiência do uso da rede, permitindo otimizações com as configurações batch_size e linger_ms , que controlam a quantidade e a latência dos lotes, respectivamente. A configuração acks='all' garante que as mensagens sejam armazenadas de forma durável, o que é essencial para dados de log importantes.

producer = KafkaProducer(
   bootstrap_servers=['localhost:9092'],  # Specifies the Kafka server to connect
   value_serializer=lambda x: json.dumps(x).encode('utf-8'),  # Serializes data as JSON and encodes it to UTF-8 before sending
   batch_size=16384,     # Sets the maximum batch size in bytes (here, 16 KB) for buffered messages before sending
   linger_ms=10,         # Sets the maximum delay (in milliseconds) before sending the batch
   acks='all'            # Specifies acknowledgment level; 'all' ensures message durability by waiting for all replicas to acknowledge
)


def generate_log_message():
   levels = ["INFO", "WARNING", "ERROR", "DEBUG"]
   messages = [
       "User login successful",
       "User login failed",
       "Database connection established",
       "Database connection failed",
       "Service started",
       "Service stopped",
       "Payment processed",
       "Payment failed"
   ]
   log_entry = {
       "level": random.choice(levels),
       "message": random.choice(messages),
       "timestamp": time.time()
   }
   return log_entry

def send_log_batches(topic, num_batches=5, batch_size=10):
   for i in range(num_batches):
       logger.info(f"Sending batch {i + 1}/{num_batches}")
       for _ in range(batch_size):
           log_message = generate_log_message()
           producer.send(topic, value=log_message)
       producer.flush()


if __name__ == "__main__":
   topic = "logs"
   send_log_batches(topic)
   producer.close()

Ao iniciar o produtor, as mensagens são enviadas em lotes para o tópico, conforme mostrado abaixo:

INFO:kafka.conn:Set configuration …
INFO:log_producer:Sending batch 1/5 
INFO:log_producer:Sending batch 2/5
INFO:log_producer:Sending batch 3/5
INFO:log_producer:Sending batch 4/5

Consumo e indexação de dados com o Kafka Consumer.

O consumidor foi projetado para processar mensagens de forma eficiente, consumindo lotes do tópico de logs e indexando-os no Elasticsearch. Com auto_offset_reset='latest', garante-se que o consumidor comece a processar as mensagens mais recentes, ignorando as mais antigas, e max_poll_records=10 limita o lote a 10 mensagens. Com fetch_max_wait_ms=2000, o consumidor espera até 2 segundos para acumular mensagens suficientes antes de processar o lote.

Em seu loop principal, o consumidor consome mensagens de log, processa e indexa cada lote no Elasticsearch, garantindo a ingestão contínua de dados.

consumer = KafkaConsumer(
   'logs',                               
   bootstrap_servers=['localhost:9092'],
   auto_offset_reset='latest',            # Ensures reading from the latest offset if the group has no offset stored
   enable_auto_commit=True,               # Automatically commits the offset after processing
   group_id='log_consumer_group',         # Specifies the consumer group to manage offset tracking
   max_poll_records=10,                   # Maximum number of messages per batch
   fetch_max_wait_ms=2000                 # Maximum wait time to form a batch (in ms)
)

def create_bulk_actions(logs):
   for log in logs:
       yield {
           "_index": "logs",
           "_source": {
               'level': log['level'],
               'message': log['message'],
               'timestamp': log['timestamp']
           }
       }

if __name__ == "__main__":
   try:
       print("Starting message processing…")
       while True:

           messages = consumer.poll(timeout_ms=1000)  # Poll receive messages

           # process each batch messages
           for _, records in messages.items():
               logs = [json.loads(record.value) for record in records]
               bulk_actions = create_bulk_actions(logs)
               response = helpers.bulk(es, bulk_actions)
               print(f"Indexed {response[0]} logs.")
   except Exception as e:
       print(f"Erro: {e}")
   finally:
       consumer.close()
       print(f"Finish")

Visualizando dados no Kibana

Com o Kibana, podemos explorar e validar os dados ingeridos do Kafka e indexados no Elasticsearch. Ao acessar as Ferramentas de Desenvolvimento no Kibana, você pode visualizar as mensagens indexadas e confirmar se os dados estão conforme o esperado. Por exemplo, se o nosso produtor Kafka enviar 5 lotes de 10 mensagens cada, devemos ver um total de 50 registros no índice.

Para verificar os dados, você pode usar a seguinte consulta na seção Ferramentas de Desenvolvimento :

GET /logs/_search
{
  "query": {
    "match_all": {}
  }
}

Resposta.

Além disso, o Kibana oferece a capacidade de criar visualizações e painéis que podem ajudar a tornar a análise mais intuitiva e interativa. Abaixo, você pode ver alguns exemplos dos painéis e visualizações que criamos, os quais ilustram os dados em vários formatos, aprimorando nossa compreensão das informações processadas.

Ingestão de dados com Kafka Connect

O Kafka Connect é um serviço projetado para facilitar a integração entre fontes de dados e destinos (sinks), como bancos de dados ou sistemas de arquivos. Ele opera com conectores predefinidos que gerenciam a movimentação de dados automaticamente. Em nosso caso, o Elasticsearch funciona como o coletor de dados.

Ao usar o Kafka Connect, podemos simplificar o processo de ingestão de dados, eliminando a necessidade de implementar manualmente o fluxo de trabalho de ingestão de dados no Elasticsearch. Com o conector apropriado, o Kafka Connect permite que os dados enviados para um tópico do Kafka sejam indexados diretamente no Elasticsearch com configuração mínima e sem necessidade de codificação adicional.

Trabalhando com o Kafka Connect

Para implementar o Kafka Connect, adicionaremos o serviço kafka-connect à nossa configuração do Docker Compose. Uma parte fundamental dessa configuração é a instalação do conector Elasticsearch, que ficará responsável pela indexação dos dados.

Após configurar o serviço e criar o contêiner do Kafka Connect, será necessário um arquivo de configuração para o conector do Elasticsearch. Este arquivo define parâmetros essenciais, tais como:

• connection.urlURL de conexão para o Elasticsearch.

• topicsO tópico do Kafka que o conector irá monitorar (neste caso, "logs").

• type.nameTipo de documento no Elasticsearch (normalmente _doc).

• value.converterConverte mensagens do Kafka para o formato JSON.

• value.converter.schemas.enableEspecifica se o esquema deve ser incluído.

• schema.ignore e key.ignore: Configurações para ignorar esquemas e chaves do Kafka durante a indexação.

Abaixo está o comando curl para criar o conector Elasticsearch no Kafka Connect:

curl --location '{{url}}/connectors' \
--header 'Content-Type: application/json' \
--data '{
    "name": "elasticsearch-sink-connector",
    "config": {
        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
        "topics": "logs",
        "connection.url": "http://elasticsearch:9200",
        "type.name": "_doc",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": "false",
        "schema.ignore": "true",
        "key.ignore": "true"
    }
}'

Com essa configuração, o Kafka Connect começará automaticamente a ingerir os dados enviados para o tópico "logs" e a indexá-los no Elasticsearch. Essa abordagem permite a ingestão e indexação de dados totalmente automatizadas, sem a necessidade de codificação adicional, simplificando assim todo o processo de integração.

Conclusão

A integração do Kafka e do Elasticsearch cria um poderoso pipeline para ingestão e análise de dados em tempo real. Este guia fornece uma abordagem fundamental para a construção de uma arquitetura robusta de ingestão de dados, com visualização e análise integradas no Kibana, pronta para se adaptar a requisitos mais complexos no futuro.

Além disso, o uso do Kafka Connect torna a integração entre o Kafka e o Elasticsearch ainda mais simplificada, eliminando a necessidade de código adicional para processar e indexar dados. O Kafka Connect permite que os dados enviados para um tópico específico sejam indexados automaticamente no Elasticsearch com configuração mínima.