Guía rápida para Kafka Connector

Overview

Esta guía muestra cómo configurar el Kafka Connector de MongoDB para enviar datos entre MongoDB y Apache Kafka.

Después de completar esta guía, deberías entender cómo usar la REST API de Kafka Connect para configurar conectores de Kafka de MongoDB para leer datos de MongoDB y escribirlos en un tema de Kafka, así como para leer datos de un tema de Kafka y escribirlos en MongoDB.

Para completar los pasos de esta guía, debes descargar y trabajar en un sandbox, un entorno de desarrollo de contenedores que incluye los servicios necesarios para compilar una muestra de pipeline de datos.

Lee las siguientes secciones para configurar tu área de pruebas y la pipeline de datos de muestra.

Instala los paquetes requeridos

Descarga e instala los siguientes paquetes:

• Docker

• Git

El entorno sandbox utiliza Docker por conveniencia y coherencia. Para más información sobre las opciones de implementación de Apache Kafka, consulte los siguientes recursos:

• Inicio Rápido Oficial de Apache Kafka

• Guía oficial de instalación de Confluent Platform

Descarga el Sandbox

Creamos un entorno aislado que incluye los servicios que necesitas en este tutorial para construir tu pipeline de datos de muestra.

Para descargar el sandbox, clona el repositorio del tutorial a tu entorno de desarrollo. Luego navega al directorio que corresponde al tutorial de introducción rápida. Si utilizas bash o un shell similar, usa los siguientes comandos:

git clone https://github.com/mongodb-university/kafka-edu.git
cd kafka-edu/docs-examples/mongodb-kafka-base/

Iniciar el Sandbox

El sandbox inicia los siguientes servicios en contenedores Docker:

• MongoDB, configurado como un set de réplicas

• Apache Kafka

• Kafka Connect con el Kafka Connector de MongoDB instalado

• Apache Zookeeper, que gestiona la configuración de Apache Kafka

Para iniciar el entorno de pruebas, ejecuta el siguiente comando desde el directorio del tutorial:

docker compose -p mongo-kafka up -d --force-recreate

Cuando inicies el sandbox, Docker descargará todas las imágenes que necesite para ejecutarse.

Después de que Docker descargue y compile las imágenes, debería ver la siguiente salida en su entorno de desarrollo:

...
Creating zookeeper ... done
Creating broker    ... done
Creating schema-registry ... done
Creating connect         ... done
Creating rest-proxy      ... done
Creating mongo1          ... done
Creating mongo1-setup    ... done

Agregar conectores

Para completar el pipeline de datos de la muestra, debe agregar conectores a Kafka Connect para transferir datos entre Kafka Connect y MongoDB. Agregue un conector fuente para transferir datos desde MongoDB a Apache Kafka. Agrega un conector de sink para transferir datos de Apache Kafka a MongoDB.

Para agregar conectores en el entorno sandbox, primero inicia un shell bash interactivo en tu contenedor Docker usando el siguiente comando:

docker exec -it mongo1 /bin/bash

Después de que se inicie su sesión de shell, debería ver el siguiente mensaje:

MongoDB Kafka Connector Sandbox $

curl -X POST \
     -H "Content-Type: application/json" \
     --data '
     {"name": "mongo-source",
      "config": {
         "connector.class":"com.mongodb.kafka.connect.MongoSourceConnector",
         "connection.uri":"mongodb://mongo1:27017/?replicaSet=rs0",
         "database":"quickstart",
         "collection":"sampleData",
         "pipeline":"[{\"$match\": {\"operationType\": \"insert\"}}, {$addFields : {\"fullDocument.travel\":\"MongoDB Kafka Connector\"}}]"
         }
     }
     ' \
     http://connect:8083/connectors -w "\n"

...
curl: (7) Failed to connect to connect port 8083: Connection refused

curl -X GET http://connect:8083/connectors

["mongo-source"]

curl -X POST \
     -H "Content-Type: application/json" \
     --data '
     {"name": "mongo-sink",
      "config": {
         "connector.class":"com.mongodb.kafka.connect.MongoSinkConnector",
         "connection.uri":"mongodb://mongo1:27017/?replicaSet=rs0",
         "database":"quickstart",
         "collection":"topicData",
         "topics":"quickstart.sampleData",
         "change.data.capture.handler": "com.mongodb.kafka.connect.sink.cdc.mongodb.ChangeStreamHandler"
         }
     }
     ' \
     http://connect:8083/connectors -w "\n"

curl -X GET http://connect:8083/connectors

["mongo-source", "mongo-sink"]

mongosh mongodb://mongo1:27017/?replicaSet=rs0

rs0 [primary] test>

use quickstart
db.sampleData.insertOne({"hello":"world"})

db.topicData.find()

[
    {
      _id: ObjectId(...),
      hello: 'world',
      travel: 'MongoDB Kafka Connector'
    }
]

exit

Remover la Sandbox

Para conservar los recursos en su entorno de desarrollo, remueva el entorno de pruebas.

Antes de remover el sandbox, sal de la sesión de shell en tu contenedor Docker ejecutando el siguiente comando:

exit

Puedes optar por remover tanto los contenedores de Docker como las imágenes, o únicamente los contenedores. Si remueves los contenedores e imágenes, tendrás que volver a descargarlos para reiniciar tu sandbox, que tiene un tamaño aproximado de 2,4 GB. Si remueves exclusivamente los contenedores, puedes reutilizar las imágenes y evitar descargar la mayoría de los archivos grandes en el pipeline de datos de muestra.

Selecciona la pestaña que corresponde a la tarea de eliminación que deseas ejecutar.

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

docker-compose -p mongo-kafka down

Próximos pasos

Para aprender cómo instalar el MongoDB Kafka Connector, consulta la guía Instalar el MongoDB Kafka Connector.

Para obtener más información sobre cómo procesar y mover datos de Apache Kafka a MongoDB, consulta la guía del Sink Connector.

Para obtener más información sobre cómo procesar y transferir datos de MongoDB a Apache Kafka, consulta la guía Conector de origen.