SparkでKafkaを経由して外部のデータを読取りします「KafkaコネクトMQTT」

本章では、Kafkaコネクタライブラリを使ってMQTTのトピックにアクセスしてMQTTに生産されたメッセージをKafkaのトピックに書き込む処理を説明します。
Kafkaコネクタを使う前に以下を確認してください。

- Kafkaクラスタの動作環境がセットアップされていたかどうかを確認してください。
- MQTTブローカーを準備しておく必要があります。


#１．connect-standalone.propertiesの編集

KafkaでMQTTへコネクトする時に、connect-standalone.propertiesを使います。
以下のように、connect-standalone.propertiesを編集することが必要です。

```
cd /usr/local/lib/kafka/config
vi connect-standalone.properties
# 編集
bootstrap.servers=192.168.0.97:9092,192.168.0.98:9092,192.168.0.99:9092
offset.storage.file.filename=/var/lib/connect.offsets
```

**オプションの使い方**

オプション|解説
:--|:--
bootstrap.servers<img width="50px" />|各ノードのIPで設定します。ポートは変わらず。
offset.storage.file.filename|Offset情報格納ファイルです。<br>このファイルは一度、空のファイルで作成しないといけません。<br>書込み権限付与も必須です。

#２．Kafkaコネクタライブラリのセットアップ
Kafkaコネクタを利用してMQTTブローカーかMQTTのセンサデータをApache Kafkaに流し込んでいきます。
このKafkaコネクタはApache Kafkaの正式なコネクタではなく、コミュニティからのものです。
Githubからソースコード見るときには、以下のリンクを利用してください。
[kafka-connect-mqttリソースのダウンロード先](https://github.com/evokly/kafka-connect-mqtt)

KafkaコネクMQTTのライブラリーをKafkaのlibsにコピーします。
対象Jars： 
> kafka-connect-mqtt-1.0-SNAPSHOT.jar
> org.eclipse.paho.client.mqttv3-1.0.2.jar

コピー先： 
> /usr/local/lib/kafka/libs

KafkaコネクMQTT用の構成ファイルをKafkaのconfigにコピーします。
対象File：
> mqtt.properties

コピー先： 
> /usr/local/lib/kafka/config

**mqtt.propertiesの内容**

> &#35;&#35;
> &#35; Basic
> &#35;&#35;
> name=mqtt
> connector.class=com.evokly.kafka.connect.mqtt.MqttSourceConnector
> tasks.max=1
> &#35;&#35;
> &#35;Settings
> &#35;&#35;
> &#35; Where to put processed messages - default to `mqtt`
> kafka.topic=topic-mqtt-kafka
> &#35; What client id to use - defaults to `null` which means random client_id
> mqtt.client_id=mqtt-kafka-99
> &#35; Use clean session in connection? - default `true`
> mqtt.clean_session=true
> &#35; What mqtt connection timeout to use - defaults to `30` seconds
> mqtt.connection_timeout=30
> &#35; What mqtt connection keep alive to use - defaults to `60` seconds
> mqtt.keep_alive_interval=60
> &#35; Mqtt broker address to use - defaults to `tcp://localhost:1883`
> &#35; if using TLS then certs can be used
> mqtt.server_uris=tcp://192.168.0.99:1883
> &#35; Mqtt topic to listen to - defaults to `#` (wildcard - all)
> mqtt.topic=topic-mqtt
> &#35; エンコード無しようになる設定
> key.converter.schemas.enable=false
> value.converter.schemas.enable=false
> value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
> key.converter=org.apache.kafka.connect.json.JsonConverter

Kafkaコネクタライブラリのセットアップは以上で完了です。
以上の設定ができたらMQTTブローカーのトピック「topic-mqtt」->Apache Kafkaのトピック「topic-mqtt-kafka」に流し込んでくるのが出来たはずです。

#３．KafkaコネクタMQTTセンサデータ取得手順
Kafkaコネクタのセットアップされていた上で、以下の手順の通りに、
KafkaのコンシューマーMQTTブローカーから流し込んできたメッセージを受信してみましょう。
**Kafkaクラスタ起動**

```
cd /usr/local/lib/zookeeper
bin/zkServer.sh start
```
停止：zkServer.sh stop
クラスターの場合は、全ノードのZookeeperが起動必要。
**Kafka起動**

```
#Zookeeper起動
cd /usr/local/lib/zookeeper
bin/zkServer.sh start
#Kafkaの起動
cd /usr/local/lib/kafka
bin/kafka-server-start.sh -daemon config/server.properties
```
**topic-mqtt-topic作成（既存の場合は不要）**

```
bin/kafka-topics.sh --create --topic topic-mqtt-kafka --replication-factor 1 --partitions 1 --zookeeper 192.168.0.99:2181
```

**Kafka コネクタMQTTに接続**

```
bin/connect-standalone.sh config/connect-standalone.properties config.mqtt.properties
```

**mosquittoより送信(MQTT送信に当たり)**

```
cd /usr/bin/
mosquitto -c /etc/mosquitto/mosquitto.conf -v
mosquitto_pub -h 192.168.0.99 -p 1883 -t topic-mqtt -m "Hello Kafka!"
```
**Kafka上で受信**

```
cd /usr/local/lib/kafka
bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.99:9092 --topic topic-mqtt-karka --from-beginning
```
以下の表示が確認しできれば受信済になります。
('Hello Kafka!')

-----
**&#9734;&#9733;&#9734; 次の文章へ &#9734;&#9733;&#9734;**
 [SparkでKafkaを経由して外部のデータを読取りします「SparkStreamingでデータ取得」](https://qiita.com/gakuseikai/items/25894099e058d7a6af8b)