Conectorul MongoDB Sink pentru Kafka Connect

Scris de Denis Nuțiu September 19, 2021

6 minute de citit
10 exemple de cod în YAML
2 imagini

Disponibil și în English

Pe această pagină

Salut 👋,

În acest articol vom construi un pipeline de date care conectează Kafka la MongoDB.

Pe scurt, vom adăuga un MongoDB Sink connector într-un cluster Kafka Connect și vom rula o instanță MongoDB în Docker pentru a testa connectorul.

Citind acest articol sper că vei învăța

Cum să instalezi connectorul MongoDB în Kafka Connect
Cum să configurezi connectorul MongoDB
Cum să creezi topicuri în Kafka folosind uneltele Confluent
Cum să monitorizezi Kafka Connect folosind JConsole.

Să începem!

Rularea MongoDB cu Docker Compose 🚢

Confluent ne oferă un fișier docker-compose care conține deja tot ce avem nevoie, în afară de câteva mici modificări.

Te rugăm să descarci următorul fișier și să-l deschizi în editorul tău preferat:

---version: '2'services:  zookeeper:    image: confluentinc/cp-zookeeper:6.2.0    hostname: zookeeper    container_name: zookeeper    ports:      - "2181:2181"    environment:      ZOOKEEPER_CLIENT_PORT: 2181      ZOOKEEPER_TICK_TIME: 2000  broker:    image: confluentinc/cp-kafka:6.2.0    hostname: broker    container_name: broker    depends_on:      - zookeeper    ports:      - "29092:29092"      - "9092:9092"      - "9101:9101"    environment:      KAFKA_BROKER_ID: 1      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0      KAFKA_JMX_PORT: 9101      KAFKA_JMX_HOSTNAME: localhost  schema-registry:    image: confluentinc/cp-schema-registry:6.2.0    hostname: schema-registry    container_name: schema-registry    depends_on:      - broker    ports:      - "8081:8081"    environment:      SCHEMA_REGISTRY_HOST_NAME: schema-registry      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081  connect:    image: cnfldemos/kafka-connect-datagen:0.5.0-6.2.0    hostname: connect    container_name: connect    depends_on:      - broker      - schema-registry    ports:      - "8083:8083"    environment:      CONNECT_BOOTSTRAP_SERVERS: 'broker:29092'      CONNECT_REST_ADVERTISED_HOST_NAME: connect      CONNECT_REST_PORT: 8083      CONNECT_GROUP_ID: compose-connect-group      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"      CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR  ksqldb-server:    image: confluentinc/cp-ksqldb-server:6.2.0    hostname: ksqldb-server    container_name: ksqldb-server    depends_on:      - broker      - connect    ports:      - "8088:8088"    environment:      KSQL_CONFIG_DIR: "/etc/ksql"      KSQL_BOOTSTRAP_SERVERS: "broker:29092"      KSQL_HOST_NAME: ksqldb-server      KSQL_LISTENERS: "http://0.0.0.0:8088"      KSQL_CACHE_MAX_BYTES_BUFFERING: 0      KSQL_KSQL_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"      KSQL_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"      KSQL_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"      KSQL_KSQL_CONNECT_URL: "http://connect:8083"      KSQL_KSQL_LOGGING_PROCESSING_TOPIC_REPLICATION_FACTOR: 1      KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: 'true'      KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: 'true'  ksqldb-cli:    image: confluentinc/cp-ksqldb-cli:6.2.0    container_name: ksqldb-cli    depends_on:      - broker      - connect      - ksqldb-server    entrypoint: /bin/sh    tty: true  ksql-datagen:    image: confluentinc/ksqldb-examples:6.2.0    hostname: ksql-datagen    container_name: ksql-datagen    depends_on:      - ksqldb-server      - broker      - schema-registry      - connect    command: "bash -c 'echo Waiting for Kafka to be ready... && \                       cub kafka-ready -b broker:29092 1 40 && \                       echo Waiting for Confluent Schema Registry to be ready... && \                       cub sr-ready schema-registry 8081 40 && \                       echo Waiting a few seconds for topic creation to finish... && \                       sleep 11 && \                       tail -f /dev/null'"    environment:      KSQL_CONFIG_DIR: "/etc/ksql"      STREAMS_BOOTSTRAP_SERVERS: broker:29092      STREAMS_SCHEMA_REGISTRY_HOST: schema-registry      STREAMS_SCHEMA_REGISTRY_PORT: 8081  rest-proxy:    image: confluentinc/cp-kafka-rest:6.2.0    depends_on:      - broker      - schema-registry    ports:      - 8082:8082    hostname: rest-proxy    container_name: rest-proxy    environment:      KAFKA_REST_HOST_NAME: rest-proxy      KAFKA_REST_BOOTSTRAP_SERVERS: 'broker:29092'      KAFKA_REST_LISTENERS: "http://0.0.0.0:8082"      KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'

Aplică următoarele modificări în fișier: poți înlocui blocul connect și poți adăuga blocul mongodb la sfârșitul fișierului.

    connect:    image: cnfldemos/kafka-connect-datagen:0.5.0-6.2.0    hostname: connect    container_name: connect    depends_on:      - broker      - schema-registry    ports:      - "8083:8083"      - "9102:9102"    environment:      CONNECT_BOOTSTRAP_SERVERS: 'broker:29092'      CONNECT_REST_ADVERTISED_HOST_NAME: connect      CONNECT_REST_PORT: 8083      CONNECT_GROUP_ID: compose-connect-group      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.storage.StringConverter      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"      CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR      KAFKA_JMX_PORT: 9102      KAFKA_JMX_HOSTNAME: localhost  mongodb:    image: mongo:4.2-rc-bionic    hostname: mongodb    container_name: mongodb    depends_on:      - broker      - connect    ports:      - 27017:27017

Rulează docker-compose up pentru a porni toate serviciile și verifică că totul rulează cu docker ps:

CONTAINER ID   IMAGE                                         COMMAND                  CREATED          STATUS         PORTS                       NAMES95165f0156f4   confluentinc/cp-ksqldb-cli:6.2.0              "/bin/sh"                37 minutes ago   Up 37 minutes                           ksqldb-cliecc4cde0f30b   confluentinc/ksqldb-examples:6.2.0            "bash -c 'echo Waiti…"   37 minutes ago   Up 37 minutes                           ksql-datagen962204b34543   mongo:4.2-rc-bionic                           "docker-entrypoint.s…"   37 minutes ago   Up 37 minutes             0.0.0.0:27017->27017/tcp, :::27017->27017/tcp                       mongodbc950f33f501a   confluentinc/cp-ksqldb-server:6.2.0           "/etc/confluent/dock…"   37 minutes ago   Up 37 minutes             0.0.0.0:8088->8088/tcp, :::8088->8088/tcp                       ksqldb-server3527577701d3   confluentinc/cp-kafka-rest:6.2.0              "/etc/confluent/dock…"   37 minutes ago   Up 37 minutes             0.0.0.0:8082->8082/tcp, :::8082->8082/tcp                       rest-proxyca69f204f4bb   cnfldemos/kafka-connect-datagen:0.5.0-6.2.0   "/etc/confluent/dock…"   37 minutes ago   Up 31 minutes (healthy)   0.0.0.0:8083->8083/tcp, :::8083->8083/tcp, 0.0.0.0:9102->9102/tcp, :::9102->9102/tcp, 9092/tcp                       connectaeaea67059c3   confluentinc/cp-schema-registry:6.2.0         "/etc/confluent/dock…"   37 minutes ago   Up 37 minutes             0.0.0.0:8081->8081/tcp, :::8081->8081/tcp                       schema-registryb9a761b98a49   confluentinc/cp-kafka:6.2.0                   "/etc/confluent/dock…"   37 minutes ago   Up 37 minutes             0.0.0.0:9092->9092/tcp, :::9092->9092/tcp, 0.0.0.0:9101->9101/tcp, :::9101->9101/tcp, 0.0.0.0:29092->29092/tcp, :::29092->29092/tcp   brokerca63570b60d4   confluentinc/cp-zookeeper:6.2.0               "/etc/confluent/dock…"   37 minutes ago   Up 37 minutes             2888/tcp, 0.0.0.0:2181->2181/tcp, :::2181->2181/tcp, 3888/tcp                       zookeeper

Instalarea MongoDB Sink Connector pe Kafka Connect 🌠

Poți descărca connectorul direct de pe Github mongodb/mongo-kafka/releases/tag/r1.6.0.

Dă click pe mongodb-kafka-connect-mongodb-1.6.0.zip, apoi dezarhivează-l și copiază directorul în calea de plugin /usr/share/java, așa cum este definită în variabila de mediu CONNECT_PLUGIN_PATH: “/usr/share/java,/usr/share/confluent-hub-components”.

Pentru a-l copia poți rula:

docker cp .\mongodb-kafka-connect-mongodb-1.6.0\ connect:/usr/share/java/ docker restart connectconnect

Connect trebuie repornit pentru a prelua pluginul nou instalat. Verifică dacă pluginul connectorului a fost instalat cu succes:

➜  bin curl -s -X GET http://localhost:8083/connector-plugins | jq | head -n 20[  {    "class": "com.mongodb.kafka.connect.MongoSinkConnector",    "type": "sink",    "version": "1.6.0"  },  {    "class": "com.mongodb.kafka.connect.MongoSourceConnector",    "type": "source",    "version": "1.6.0"  },

Notă: Dacă nu ai jq instalat, poți să-l omiți.

Crearea topicurilor

Înainte de a porni connectorul, să creăm topicurile Kafka events și events.deadletter, care vor fi folosite de connector.

Pentru a crea topicurile, va trebui să descarcăm uneltele Confluent și să rulăm kafka-topics.

curl -s -O http://packages.confluent.io/archive/6.2/confluent-community-6.2.0.tar.gztar -xzf .\confluent-community-6.2.0.tar.gzcd .\confluent-6.2.0\bin\ ./kafka-topics --bootstrap-server localhost:9092 --list__consumer_offsets__transaction_state_confluent-ksql-default__command_topic_schemasdefault_ksql_processing_logdocker-connect-configsdocker-connect-offsetsdocker-connect-status./kafka-topics --bootstrap-server localhost:9092 --create --topic events --partitions 3Created topic events../kafka-topics --bootstrap-server localhost:9092 --create --topic events.deadletter --partitions 3WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues, it is best to use either, but not both.Created topic events.deadletter.

** Notă **: Ai nevoie de Java pentru a rula uneltele Confluent. Dacă ești pe Ubuntu, poți tasta sudo apt install openjdk-8-jdk.

Pornirea connectorului 🚙

Pentru a porni connectorul, este suficient un singur request POST către API-ul connectorului cu configurația connectorului.

Configurația pe care o vom folosi este:

curl --request POST \  --url http://localhost:8083/connectors \  --header 'Content-Type: application/json' \  --data '{	"name": "mongo-sink-connector",	"config": {		"connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",		"tasks.max": "1",		"topics": "events",		"connection.uri": "mongodb://mongodb:27017/my_events",		"database": "my_events",		"collection": "kafka_events",		"max.num.retries": 5,		"mongo.errors.tolerance": "all",		"mongo.errors.log.enable": true,		"errors.log.include.messages": true,		"errors.deadletterqueue.topic.name": "events.deadletter",		"errors.deadletterqueue.context.headers.enable": true,	}}'

Pe scurt, acest POST va crea un connector nou numit mongo-sink-connector, folosind clasa Java com.mongodb.kafka.connect.MongoSinkConnector, și va rula un singur task de connector care va prelua toate mesajele din topicul events și le va pune în MongoDB de la adresa mongodb://mongodb:27017/my_events, în baza de date my_events și colecția kafka_events. Înregistrările care nu pot fi scrise în baza de date vor fi plasate într-un dead letter topic numit events.deadletter. După părerea mea, este mai bine decât să le ignorăm, deoarece putem inspecta topicul pentru a vedea ce a mers greșit.

Pentru a verifica dacă connectorul rulează, poți obține statusul primului său task cu:

➜  bin curl -s -X GET http://localhost:8083/connectors/mongo-sink-connector/tasks/0/status | jq{  "id": 0,  "state": "RUNNING",  "worker_id": "connect:8083"}

Interogarea bazei de date 🗃

Acum că cluster-ul nostru Kafka Connect rulează și este configurat, tot ce mai rămâne de făcut este să trimitem niște date de test în Kafka și să le verificăm în baza de date.

curl --request POST \  --url http://localhost:8082/topics/events \  --header 'Content-Type: application/vnd.kafka.json.v2+json' \  --data '{	"records": [		{			"key": "somekey",			"value": {				"glossary": {					"title": "example glossary",					"GlossDiv": {						"title": "S",						"GlossList": {							"GlossEntry": {								"ID": "SGML",								"SortAs": "SGML",								"GlossTerm": "Standard Generalized Markup Language",								"Acronym": "SGML",								"Abbrev": "ISO 8879:1986",								"GlossDef": {									"para": "A meta-markup language, used to create markup languages such as DocBook.",									"GlossSeeAlso": [										"GML",										"XML"									]								},								"GlossSee": "markup"							}						}					}				}			}		}	]}'

Gata! 🎉 Dacă ne conectăm acum la baza de date folosind mongosh sau orice alt client, putem interoga datele.

mongosh> use my_eventsswitched to db my_events> db.kafka_events.findOne(){  _id: ObjectId("6147242856623b0098fc756d"),  glossary: {    title: 'example glossary',    GlossDiv: {      title: 'S',      GlossList: {        GlossEntry: {          ID: 'SGML',          SortAs: 'SGML',          GlossTerm: 'Standard Generalized Markup Language',          Acronym: 'SGML',          Abbrev: 'ISO 8879:1986',          GlossDef: {            para: 'A meta-markup language, used to create markup languages such as DocBook.',            GlossSeeAlso: [ 'GML', 'XML' ]          },          GlossSee: 'markup'        }      }    }  }}

Vizualizarea metricilor JMX ale Kafka Connect

JConsole este o unealtă care poate fi folosită pentru a vedea metricile JMX expuse de Kafka Connect. Dacă ai instalat openjdk-8, ar trebui să o aibă deja.

Pornește JConsole și conectează-te la localhost:9102. Dacă primești un avertisment despre o conexiune nesigură, acceptă conexiunea și ignoră-l.

După ce ești conectat, dă click pe tab-ul MBeans și explorează 🦹‍♀️

Rezumat

La început, Kafka și Kafka Connect pot părea puțin copleșitoare. Sper că acest tutorial ți-a oferit bazele necesare ca să poți continua să te joci și să explorezi pe cont propriu.

Crearea unui mediu de joacă pentru Kafka și Connect folosind docker-compose nu este atât de complicată. Poți porni de la repo-ul confluent-cp-community, care îți oferă tot ce ai nevoie ca să începi. Cu mici modificări ale fișierului docker-compose, am pornit o instanță MongoDB și am expus metricile JMX din Kafka Connect.

Apoi, am instalat și configurat connectorul MongoDB și am confirmat că funcționează așa cum era de așteptat.

Dacă ai întrebări, spune-mi în comentarii.

Până data viitoare! 🍻