Kafka Connector 빠른 시작

개요

이 가이드에서는 MongoDB와 Apache Kafka 간에 데이터를 전송하기 위해 MongoDB Kafka Connector를 구성하는 방법을 설명합니다.

이 가이드를 완료하고 나면 Kafka Connect REST API를 사용하여 MongoDB에서 데이터를 읽고 이를 Kafka 주제에 쓰거나, Kafka 주제에서 데이터를 읽고 이를 MongoDB 쓰기 위해 MongoDB Kafka Connector를 구성하는 방법을 알 수 있습니다.

이 가이드의 단계를 모두 완료하려면 샘플 데이터 파이프라인을 구축하는데 필요한 서비스가 포함된, 컨테이너화된 개발 환경인 샌드박스를 다운로드하여 작업해야 합니다.

샌드박스 및 샘플 데이터 파이프라인을 설정하려면 다음 섹션을 읽어보세요.

필수 패키지 설치

다음 패키지를 다운로드하여 설치하세요.

• Docker

• Git

샌드박스는 편의성과 일관성을 위해 Docker를 사용합니다. Apache Kafka의 배포 옵션에 대해 자세히 알아보려면 다음 리소스를 참조하세요.

• 공식 Apache Kafka 빠른 시작

• 공식 Confluent 플랫폼 설치 가이드

Sandbox 다운로드

이 튜토리얼에서는 샘플 데이터 파이프라인을 구축하는 데 필요한 서비스가 포함된 샌드박스를 만들었습니다.

샌드박스를 다운로드하려면 튜토리얼 저장소를 사용 중인 개발 환경에 복제하세요. 그런 다음 빠른 시작 튜토리얼에 해당하는 디렉토리로 이동합니다. Bash 또는 유사한 셸을 사용하는 경우 다음 명령을 사용하세요.

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

샌드박스 시작

샌드박스는 Docker 컨테이너에서 다음 서비스를 시작합니다.

• 복제본 세트로 구성된 MongoDB

• Apache Kafka

• MongoDB Kafka Connector가 설치된 Kafka Connect

• Apache Kafka의 구성을 관리하는 Apache Zookeeper

샌드박스를 시작하려면 튜토리얼 디렉토리에서 다음 명령을 실행합니다.

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

샌드박스를 시작하면 Docker는 실행에 필요한 모든 이미지를 다운로드합니다.

Docker가 이미지를 다운로드하고 빌드하면 개발 환경에 다음 출력이 표시됩니다.

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

Connector 추가

샘플 데이터 파이프라인을 완료하려면 Kafka Connect에 connector 추가하여 Kafka Connect와 MongoDB 간에 데이터를 전송해야 합니다. 소스 connector를 추가하여 MongoDB에서 Apache Kafka로 데이터를 전송합니다. 싱크 connector를 추가하여 Apache Kafka에서 MongoDB로 데이터를 전송합니다.

샌드박스에 커넥터를 추가하려면 먼저 다음 명령을 사용하여 Docker 컨테이너에서 대화형 bash 셸을 시작합니다.

docker exec -it mongo1 /bin/bash

셸 세션이 시작되면 다음 프롬프트가 표시됩니다.

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

샌드박스 제거

개발 환경의 리소스를 절약하려면 샌드박스를 제거하세요.

샌드박스를 제거하기 전에 다음 명령을 실행하여 Docker 컨테이너의 셸 세션을 종료하세요.

exit

Docker container와 이미지를 모두 제거하거나 container만 제거하도록 선택할 수 있습니다. container와 이미지를 제거할 경우 다시 다운로드해 약 2.4GB 크기의 샌드박스를 다시 시작해야 합니다. container만 제거할 경우 이미지는 재사용할 수 있으며 샘플 데이터 파이프라인에서 대용량 파일 대부분을 다운로드하지 않아도 됩니다.

실행할 제거 작업에 해당하는 탭을 선택합니다.

다음 단계

MongoDB Kafka Connector를 설치하는 방법을 알아보려면 MongoDB Kafka Connector 설치 가이드를 참조하세요.

Apache Kafka에서 MongoDB로 데이터를 처리하고 이동하는 방법에 대해 자세히 알아보려면 싱크 커넥터 가이드를 참조하세요.

MongoDB에서 Apache Kafka로 데이터를 처리하고 이동하는 방법에 대해 자세히 알아보려면 소스 커넥터 가이드를 참조하세요.