7.2. 分散モードでの Kafka Connect

クラスターの実行

1. すべての Kafka Connect ワーカーノードで /opt/kafka/config/connect-distributed.properties Kafka Connect 設定ファイルを編集します。

   • bootstrap.server オプションを設定して、Kafka ブローカーを示すようにします。
   • group.id オプションを設定します。
   • config.storage.topic オプションを設定します。
   • offset.storage.topic オプションを設定します。

   • status.storage.topic オプションを設定します。

     以下に例を示します。

     bootstrap.servers=kafka0.my-domain.com:9092,kafka1.my-domain.com:9092,kafka2.my-domain.com:9092
     group.id=my-group-id
     config.storage.topic=my-group-id-configs
     offset.storage.topic=my-group-id-offsets
     status.storage.topic=my-group-id-status

     Copy to Clipboard Toggle word wrap

2. すべての Kafka Connect ワーカーノードで /opt/kafka/config/connect-distributed.properties Kafka Connect ワーカーを起動します。

   su - kafka
   /opt/kafka/bin/connect-distributed.sh /opt/kafka/config/connect-distributed.properties

   Copy to Clipboard Toggle word wrap

3. KafkaConnect が実行されていることを確認します。

   jcmd | grep ConnectDistributed

   Copy to Clipboard Toggle word wrap

手順

1. コネクター設定で JSON ペイロードを準備します。以下に例を示します。

   {
     "name": "my-connector",
     "config": {
     "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
       "tasks.max": "1",
       "topics": "my-topic-1,my-topic-2",
       "file": "/tmp/output-file.txt"
     }
   }

   Copy to Clipboard Toggle word wrap

2. POST リクエストを <KafkaConnectAddress>:8083/connectors に送信してコネクターを作成します。以下の例では、curl を使用します。

   curl -X POST -H "Content-Type: application/json" --data @sink-connector.json http://connect0.my-domain.com:8083/connectors

   Copy to Clipboard Toggle word wrap

3. <KafkaConnectAddress>:8083/connectors に GET リクエストを送信して、コネクターがデプロイされたことを確認します。以下の例では、curl を使用します。

   curl http://connect0.my-domain.com:8083/connectors

   Copy to Clipboard Toggle word wrap

コネクターの削除

1. <KafkaConnectAddress>:8083/connectors/<ConnectorName> に GET リクエストを送信して、コネクターが存在することを確認します。以下の例では、curl を使用します。

   curl http://connect0.my-domain.com:8083/connectors

   Copy to Clipboard Toggle word wrap

2. コネクターを削除するには、DELETE リクエストを <KafkaConnectAddress>:8083/connectors に送信します。以下の例では、curl を使用します。

   curl -X DELETE http://connect0.my-domain.com:8083/connectors/my-connector

   Copy to Clipboard Toggle word wrap

3. <KafkaConnectAddress>:8083/connectors に GET リクエストを送信して、コネクターが削除されたことを確認します。以下の例では、curl を使用します。

   curl http://connect0.my-domain.com:8083/connectors

   Copy to Clipboard Toggle word wrap