Guía de inicio rápido del conector Kafka

Overview

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

Después de completar esta guía, debería comprender cómo usar la API REST de Kafka Connect para configurar los conectores Kafka de MongoDB para leer datos de MongoDB y escribirlos en un tema de Kafka, y 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.

Lea las siguientes secciones para configurar su entorno aislado y su canalización de datos de muestra.

Instalar los paquetes necesarios

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 sandbox que incluye los servicios que necesita en este tutorial para construir su canalización 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 administra la configuración de Apache Kafka

Para iniciar el sandbox, ejecute 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 la canalización de datos de ejemplo, debe agregar conectores a Kafka Connect para transferir datos entre Kafka Connect y MongoDB. Agregue un conector de origen para transferir datos de MongoDB a Apache Kafka. Agregue un conector de destino 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 eliminar el entorno limitado, salga de la sesión de shell en su 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, consulte la guía del conector Sink.

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.