Começando com o conector do coletor MongoDB Kafka

Concluir a configuração do tutorial

Conclua as etapas daConfiguração do tutorial doKafka Connector para iniciar o Kafka Connect da Confluent e o ambiente do MongoDB .

Configurar o conector do coletor

Crie uma sessão de shell interativa no tutorial Docker Container usando o seguinte comando:

docker exec -it mongo1 /bin/bash

Crie um arquivo de configuração de origem denominado simplesink.json com o seguinte comando:

nano simplesink.json

Cole as seguintes informações de configuração no arquivo e salve suas alterações:

{
  "name": "mongo-tutorial-sink",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
    "topics": "Tutorial2.pets",
    "connection.uri": "mongodb://mongo1",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": false,
    "database": "Tutorial2",
    "collection": "pets"
  }
}

Execute o seguinte comando no shell para iniciar o conector do coletor usando o arquivo de configuração criado:

cx simplesink.json

curl -X POST -H "Content-Type: application/json" -d @simplesink.json http://connect:8083/connectors -w "\n"

Execute o seguinte comando na shell para verificar o status dos conectores:

status

Se o conector do coletor tiver sido iniciado com sucesso, você verá a seguinte saída:

Kafka topics:
...
The status of the connectors:
sink  |  mongo-tutorial-sink  |  RUNNING  |  RUNNING  |  com.mongodb.kafka.connect.MongoSinkConnector
Currently configured connectors
[
"mongo-tutorial-sink"
]
...

Grave dados em um tópico do Kafka

Na mesma shell, crie um roteiro do Python para gravar dados em um tópico do Kafka.

nano kafkawrite.py

Cole o código abaixo no arquivo e salve as alterações:

from kafka import KafkaProducer
import json
from json import dumps
p = KafkaProducer(bootstrap_servers = ['broker:29092'], value_serializer = lambda x:dumps(x).encode('utf-8'))
data = {'name': 'roscoe'}
p.send('Tutorial2.pets', value = data)
p.flush()

Execute o roteiro do Python:

python3 kafkawrite.py

Visualizar os dados na collection MongoDB

Na mesma shell, conecte ao MongoDB utilizando o mongosh, a shell MongoDB executando o seguinte comando:

mongosh "mongodb://mongo1"

Depois de se conectar com sucesso, você deverá ver o seguinte prompt do shell do MongoDB:

rs0 [direct: primary] test>

No prompt, digite os seguintes comandos para recuperar todos os documentos no namespace MongoDB do Tutorial2.pets:

use Tutorial2
db.pets.find()

O resultado deve ver o seguinte documento retornado:

{ _id: ObjectId("62659..."), name: 'roscoe' }

Saia da shell MongoDB inserindo o comando exit.

(Opcional) Interromper os containers do Docker

Depois de concluir este tutorial, libere recursos em seu computador interrompendo ou removendo os ativos do Docker. Você pode remover os containers e as imagens do Docker, ou somente os containers. Se você remover os containers e as imagens, será necessário baixá-los novamente para reiniciar seu ambiente de desenvolvimento do Conector Kafka MongoDB, que tem aproximadamente 2,4 GB. Se você remover somente os containers, poderá reutilizar as imagens sem precisar baixar a maioria dos arquivos grandes na pipeline de dados de amostra.

Selecione a aba que corresponde à tarefa de remoção que deseja executar.

docker-compose -p mongo-kafka down --rmi all

docker-compose -p mongo-kafka down

Para reiniciar os containers, siga as mesmas etapas necessárias para iniciá-los na configuração do tutorial.