Kafka Connectorクイック スタート

Overview

このガイドでは、MongoDB Kafka Connector を構成して、MongoDB と Apache Kafka 間でデータを送信する方法について説明します。

このガイドを完了すると、 Kafka Connect REST APIを使用してMongoDB Kafka Connector を構成して、 MongoDBからデータを読み取り、 Kafkaトピックに書込み、またKafkaトピックからデータを読み取り、 MongoDBに書込む方法を理解する必要があります。

このガイドの手順を完了するには、 サンプル データ パイプライン の構築に必要なサービスを含むコンテナ化された開発環境である サンドボックス をダウンロードして操作する必要があります。

サンドボックスとサンプル データ パイプラインを設定するには、次のセクションをお読みください。

必要なパッケージをインストールします

次のパッケージをダウンロードしてインストールします。

• Docker

• Git

サンドボックスは、便宜上と一貫性のために Docker を使用します。 Apache Kafka の配置オプションの詳細については、次のリソースを参照してください。

• 公式 Apache Kafka クイック スタート

• Confluent プラットフォームの公式インストール ガイド

サンドボックスをダウンロード

このチュートリアルでは、サンプル データ パイプラインを構築するために必要なサービスを含むサンドボックスを作成しました。

サンドボックスをダウンロードするには、チュートリアル リポジトリを開発環境にクローンします。 次に、クイックスタート チュートリアルに対応するディレクトリに移動します。 bash または同様の shell を使用する場合は、次のコマンドを使用します。

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 Zooker

サンドボックスを起動するには、チュートリアル ディレクトリから次のコマンドを実行します。

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

コネクターの追加

サンプル データ パイプラインを完了するには、 Kafka Connect と MongoDB 間でデータを転送するためのコネクターを Kafka Connect に追加する必要があります。 MongoDB から Apache Kafka にデータを転送するためのソース コネクタを追加します。 Apache Kafka から MongoDB にデータを転送するためのSink Connectorを追加します。

サンドボックスにコネクターを追加するには、まず次のコマンドを使用して コンテナ内でインタラクティブshell Dockerを起動します。

docker exec -it mongo1 /bin/bash

shell セッションを開始すると、次のプロンプトが表示されます。

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 コンテナ内の shell セッションを終了します。

exit

Docker コンテナとイメージの両方を削除することも、 コンテナのみを削除することも選択できます。 コンテナとイメージを削除した場合、サイズが約 2.4 GB のサンドボックスを再起動するには、それらを再度ダウンロードする必要があります。 コンテナのみを削除すると、イメージを再利用し、サンプル データ パイプライン内の大きなファイルのほとんどをダウンロードしないでください。

実行する削除タスクに対応するタブを選択します。

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

docker-compose -p mongo-kafka down

次のステップ

MongoDB Kafka Connector をインストールする方法については、「 MongoDB Kafka Connector のインストール」のガイドを参照してください。

Apache Kafka から MongoDB にデータを処理して移動する方法の詳細については、 Sink Connectorのガイド を参照してください。

MongoDBからApache Kafkaにデータを処理して移動する方法の詳細については、 「ソースConnector 」 のガイドを参照してください。