Introducción al conector de sumidero de Kafka de MongoDB

Completar la configuración del tutorial

Complete los pasos en el Tutorial del conector de Kafka Configuración para iniciar el entorno de Confluent Kafka Connect y MongoDB.

Configurar el conector del sumidero

Cree una sesión de shell interactiva en el tutorial Docker Container usando el siguiente comando:

docker exec -it mongo1 /bin/bash

Cree un archivo de configuración de origen llamado simplesink.json con el siguiente comando:

nano simplesink.json

Pegue la siguiente información de configuración en el archivo y guarde los cambios:

{
  "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"
  }
}

Ejecute el siguiente comando en el shell para iniciar el conector del receptor usando el archivo de configuración que creó:

cx simplesink.json

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

Ejecute el siguiente comando en el shell para comprobar el estado de los conectores:

status

Si el conector del sumidero se inició correctamente, debería ver el siguiente resultado:

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

Escribir datos en un tema de Kafka

En el mismo shell, crea un script de Python para escribir datos en un tema de Kafka.

nano kafkawrite.py

Pegue el siguiente código en el archivo y guarde los cambios:

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()

Ejecute el script de Python:

python3 kafkawrite.py

Ver los datos en la colección MongoDB

En el mismo shell, conéctese a MongoDB usando mongosh, el shell de MongoDB ejecutando el siguiente comando:

mongosh "mongodb://mongo1"

Luego de conectarse correctamente, deberías ver el siguiente mensaje en la shell de MongoDB:

rs0 [direct: primary] test>

En el indicador, escriba los siguientes comandos para recuperar todos los documentos en el espacio de nombres Tutorial2.pets de MongoDB:

use Tutorial2
db.pets.find()

Debería ver el siguiente documento devuelto como resultado:

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

Salga del shell de MongoDB ingresando el comando exit.

(Opcional) Detener los contenedores Docker

Después de completar este tutorial, libere recursos en su equipo deteniendo o eliminando recursos de Docker. Puede eliminar tanto los contenedores como las imágenes de Docker, o solo los contenedores. Si elimina los contenedores y las imágenes, deberá descargarlos de nuevo para reiniciar su entorno de desarrollo del Conector Kafka de MongoDB, que tiene un tamaño aproximado de 2.4 GB. Si elimina solo los contenedores, puede reutilizar las imágenes y evitar la descarga de la mayoría de los archivos grandes en la canalización de datos de ejemplo.

Seleccione la pestaña que corresponde a la tarea de eliminación que desea ejecutar.

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

docker-compose -p mongo-kafka down

Para reiniciar los contenedores, siga los mismos pasos necesarios para iniciarlos en la configuración del tutorial.