Vá para o MongoDB usando conectores Kafka - Guia final do agente

Pavel Duchovny7 min read • Published Feb 08, 2022 • Updated Sep 17, 2024

Kafka Conectores Go

Avalie esse Tutorial

Go é uma linguagem moderna baseada em conceitos de compilação de código nativo e digitado, ao mesmo tempo em que sente e utiliza alguns benefícios das linguagens dinâmicas. É bastante simples de instalar e usar, pois fornece código legível e robusto para muitos casos de uso de aplicativos.

Um desses casos de uso é a criação de agentes que se reportam a uma plataforma de dados centralizada por meio de streaming. Uma abordagem amplamente aceita é comunicar os dados do agente por meio da inscrição de filas distribuídas como o Kafka. Os tópicos do Kafka podem então propagar os dados para muitas fontes diferentes, como um clusterdo MongoDB Atlas .

Ter um agente Go nos permite utilizar a mesma base de código para vários sistemas operacionais, e o fato de ele ter uma boa integração com dados e pacotes JSON, como um driver MongoDB e oConfluent Go Kafka Client, o torna um candidato tentador para o caso de uso apresentado .

Este artigo demonstrará como os dados de tamanho de arquivo em um host são monitorados a partir de um agente multiplataforma escrito em Go por meio de um cluster Kafka usando um connector de coletor hospedado pelo Confluent para o MongoDB Atlas. O MongoDB Atlas armazena os dados em uma coleção de séries temporais. O produtoMongoDB Charts é uma maneira conveniente de mostrar os dados coletados para o usuário.

Preparando o projeto Golang, cluster Kafka e MongoDB Atlas

Configurando um projeto Go

Nosso agente vai executar o Go. Portanto, você precisará instalar o software de idioma Go no seu host.

Quando esta etapa for concluída, criaremos um módulo Go para iniciar nosso projeto em nosso diretório de trabalho:

Trecho de código

Agora precisaremos adicionar a dependência do Confluent Kafka ao nosso projeto Golang:

Trecho de código

Configurar um cluster Kafka

A criação de um Confluent Kafka Cluster é feita por meio da UI do Confluent. Comece criando um cluster básico do Kafka no Confluent Cloud. Quando estiver pronto, crie um tópico para ser usado no cluster Kafka. Criei um chamado "files. "

Gere uma chave de API e um segredo de API para interagir com esse cluster Kafka. Para simplificar este tutorial, selecionei a api-key “Global Access. Para produção, recomenda-se fornecer o mínimo de permissões possível para a chave de API usada. Obtenha as chaves geradas para uso futuro.

Obtenha a connection string do cluster Kafka por meio de Visão Geral do Cluster > Configurações do Cluster > Identificação > Servidor Bootstrap para uso futuro. Os clusters básicos estão abertos à Internet e, em produção, você precisará alterar a lista de acesso para que seus hosts específicos se conectem ao cluster por meio de ACLs de cluster avançadas.

Importante: o connector Confluent exige que o Kafka cluster e o Atlas cluster sejam implantados na mesma região.

Configurando o projeto e cluster do Atlas

Crie um projeto e cluster ou utilize um Atlas cluster existente em seu projeto. Como estamos usando uma coleção de séries temporais, os clusters devem usar um 5.0versão +. Prepare seu cluster Atlas para uma conexãoAtlas de coletor Confluent. Dentro da lista de acesso do seu projeto, habilite o usuário e endereços IP relevantes dos seus IPs do conector. Os IPs da lista de acesso devem estar associados ao Atlas Sink Connector, que configuraremos em uma seção seguinte.Por fim, controle aconnection stringdo Atlas e o DNS do cluster principal. Para obter mais informações sobre a melhor proteção e a obtenção dos IPs relevantes do seu conector Confluent, leia o seguinte artigo: Conector do MongoDB Atlas Sink para a Confluent Cloud.

Adicionando lógica principal do agente

Agora que temos nosso cluster Kafka e clusters Atlas criados e preparados, podemos inicializar nosso código de agente criando um pequeno arquivo principal que monitorará meu diretório./files e capturará os nomes e tamanhos dos arquivos.Adicionei um arquivo chamadotest.txt com alguns dados para trazê-lo para ~200MB.

Vamos criar um arquivo chamado main.go e escrever uma pequena lógica que executa um loop constante com um 1 min de sono para percorrer os arquivos na pastafiles :

Trecho de código

package main

import (
   "fmt"
   "encoding/json"
   "time"
   "os"
   "path/filepath"
)

type Message struct {
  Name string
  Size float64
  Time int64
}

func samplePath (startPath string)  error {
   err := filepath.Walk(startPath,
    func(path string, info os.FileInfo, err error) error {
 
       var bytes int64
       bytes = info.Size()

var kilobytes int64
       kilobytes = (bytes / 1024)

var megabytes float64
       megabytes = (float64)(kilobytes / 1024)   // cast to type float64

var gigabytes float64
       gigabytes = (megabytes / 1024)

now := time.Now().Unix()*1000

m := Message{info.Name(), gigabytes, now}
      value, err := json.Marshal(m)

if err != nil {
         panic(fmt.Sprintf("Failed to parse JSON: %s", err))
      }

fmt.Printf("value: %v\n", string(value))
      return nil;
  })
  if err != nil {
     return err
  }
  return nil;
}

func main() {
   for {
       err := samplePath("./files");
       if err != nil {
            panic(fmt.Sprintf("Failed to run sample : %s", err))
       }
       time.Sleep(time.Minute)
   }

}

O código acima simplesmente importa módulos auxiliares para percorrer os diretórios e documentos JSON a partir dos arquivos encontrados.

Como precisamos que os dados sejam marcados com o tempo da amostra, eles são uma ótima opção para dados de séries temporais e, portanto, devem ser armazenados em uma coleção de séries temporais no Atlas. Se quiser saber mais sobre a coleta de dados de séries temporais, leia nosso artigoDados de séries temporais do MongoDB.

Podemos testar este agente executando o seguinte comando:

Trecho de código

O agente produzirá documentos JSON semelhantes ao seguinte formato:

Trecho de código

Criando um connector Confluent MongoDB para Kafka

Agora vamos criar um Kafka Sink connector para escrever os dados que chegam no tópico "files " para a coleção de séries temporais do nosso Atlas Cluster.

O Confluent Cloud tem uma integração muito popular que executa o Kafka Connector do MongoDB como uma solução hospedada integrada a seus clusters Kafka. Siga estas etapas para iniciar um sistema de connector.

A seguir estão as entradas fornecidas ao connector: Depois de configurá-lo, seguindo o guia, você eventualmente terá uma página de resumo de lançamento semelhante: Após o provisionamento, cada documento preenchido na filafiles será enviado para uma coleção de séries temporais hostMonitor.files onde o campo de data é Time e o campo de metadados é Name.

Enviando dados para o Kafka

Agora vamos editar o arquivomain.go para usar um cliente Kafka e enviar cada medida de arquivo para a fila "files ".

Adicione a biblioteca do cliente como um módulo importado:

Trecho de código

Adicione as credenciais de cloud Confluent e as informações de DNS do cluster. Substitua o<CONFLUENT-SERVER>:<CONFLUENT-PORT> encontrado na página de detalhes do Kafka Cluster e o <ACCESS-KEY> , <SECRET-KEY> gerado no Kafka Cluster:

Trecho de código

O código a seguir iniciará o produtor e produzirá uma mensagem a partir do JSON document ordenado:

Trecho de código

if err != nil {
                   panic(fmt.Sprintf("Failed to create producer: %s", err))
               }

producer.Produce(&kafka.Message{
                   TopicPartition: kafka.TopicPartition{Topic: &topic,
                       Partition: kafka.PartitionAny},
                   Value: []byte(value)}, nil)

// Wait for delivery report
               e := <-producer.Events()

message := e.(*kafka.Message)
               if message.TopicPartition.Error != nil {
                   fmt.Printf("failed to deliver message: %v\n",
                       message.TopicPartition)
               } else {
                   fmt.Printf("delivered to topic %s [%d] at offset %v\n",
                       *message.TopicPartition.Topic,
                       message.TopicPartition.Partition,
                       message.TopicPartition.Offset)
               }

producer.Close()

O arquivomain.gointeiro terá a seguinte aparência:

Trecho de código

package main

import (
   "fmt"
   "encoding/json"
   "time"
   "os"
   "path/filepath"
   "github.com/confluentinc/confluent-kafka-go/kafka")

type Message struct {
  Name string
  Size float64
  Time int64
}

const (
   bootstrapServers = "<SERVER-DNS>:<SERVER-PORT>"
   ccloudAPIKey     = "<ACCESS-KEY>"
   ccloudAPISecret  = "<SECRET-KEY>"
)

func samplePath (startPath string)  error {
  
   err := filepath.Walk(startPath,
    func(path string, info os.FileInfo, err error) error {
    if err != nil {
            return err
    }
    fmt.Println(path, info.Size())

var bytes int64
       bytes = info.Size()

var kilobytes int64
       kilobytes = (bytes / 1024)

var megabytes float64
       megabytes = (float64)(kilobytes / 1024)   // cast to type float64

var gigabytes float64
       gigabytes = (megabytes / 1024)

now := time.Now().Unix()*1000

m := Message{info.Name(), gigabytes, now}
       value, err := json.Marshal(m)

if err != nil {
         panic(fmt.Sprintf("Failed to parse JSON: %s", err))
       }

fmt.Printf("value: %v\n", string(value))

topic := "files"
               // Produce a new record to the topic...
                   producer, err := kafka.NewProducer(&kafka.ConfigMap{
                       "bootstrap.servers":       bootstrapServers,
                       "sasl.mechanisms":         "PLAIN",
                       "security.protocol":       "SASL_SSL",
                       "sasl.username":           ccloudAPIKey,
                       "sasl.password":           ccloudAPISecret})
  
                   if err != nil {
                       panic(fmt.Sprintf("Failed to create producer: %s", err))
                   }
  
                   producer.Produce(&kafka.Message{
                       TopicPartition: kafka.TopicPartition{Topic: &topic,
                           Partition: kafka.PartitionAny},
                       Value: []byte(value)}, nil)
  
                   // Wait for delivery report
                   e := <-producer.Events()
  
                   message := e.(*kafka.Message)
                   if message.TopicPartition.Error != nil {
                       fmt.Printf("failed to deliver message: %v\n",
                           message.TopicPartition)
                   } else {
                       fmt.Printf("delivered to topic %s [%d] at offset %v\n",
                           *message.TopicPartition.Topic,
                           message.TopicPartition.Partition,
                           message.TopicPartition.Offset)
                   }
  
                   producer.Close()

return nil;
})
if err != nil {
       return err
}
   return nil;
}

func main() {
   for {
       err := samplePath("./files");
       if err != nil {
            panic(fmt.Sprintf("Failed to run sample : %s", err))
       }
     time.Sleep(time.Minute)
  
   }

}

Agora, quando executarmos o agente enquanto o connector do Confluent Atlas estiver totalmente provisionado, veremos as mensagens produzidas na coleção de séries temporais hostMonitor.files:

Analisando os dados usando o MongoDB Charts

Para colocar nossos dados em uso, podemos criar alguns gráficos bonito sobre os dados de séries temporais. Em um gráfico de linha, configuramos o eixo X para usar o campo Time, o eixo Y para usar o campo Size e a série para usar o campo Name. O gráfico a seguir mostra as linhas coloridas representadas como a evolução dos diferentes tamanhos de arquivo ao longo do tempo. Agora temos um agente Go e um painel de Atlas Charts totalmente funcional para analisar as tendências de arquivos em crescimento. Essa arquitetura permite um grande espaço para extensibilidade, pois o agente Go pode ter outras funcionalidades, mais assinantes podem consumir os dados monitorados e agir sobre eles e, finalmente, o MongoDB Atlas e Atlas Charts podem ser usados por vários aplicativos e incorporados a diferentes plataformas.

Embrulhar

A criação de aplicativos Go é simples, mas oferece grandes benefícios em termos de desempenho, código entre plataformas e um grande número de bibliotecas e clientes suportados. A adição MongoDB Atlas por meio de um serviço Confluent Cloud Kafka torna a implementação de uma pilha robusta e extensível, transmite dados e armazena-os e apresenta-os de forma eficiente ao usuário final por meio de Charts.

Neste tutorial, abordamos todos os fundamentos que você precisa saber para começar a usar o Go, Kafka e o MongoDB Atlas em seus próximos projetos de streaming.

Experimente o MongoDB Atlas and Go hoje mesmo!

Avalie esse Tutorial

Relacionado

Tutorial

Implementando o MongoDB Enterprise Kubernetes Operator no Google Cloud

Jan 13, 2023 | 6 min read

Tutorial

Implantando o MongoDB em vários clusters do Kubernetes com o MongoDBMulti

Sep 05, 2023 | 11 min read

Artigo

Medindo o desempenho do MongoDB Kafka Connector

May 09, 2022 | 6 min read

Tutorial

Usando a autenticação AWS IAM com o conector MongoDB para Apache Kafka

Jul 01, 2024 | 4 min read

Sumário

Preparando o projeto Golang, cluster Kafka e MongoDB Atlas
Adicionando lógica principal do agente
Criando um connector Confluent MongoDB para Kafka
Enviando dados para o Kafka
Analisando os dados usando o MongoDB Charts
Embrulhar