PLCnext ControlとApache KafkaをMQTTで連携：エンタープライズグレード統合ガイド

技術的背景

Kafka

Apache Kafkaはデータの取り込み、保存、処理、再配布のためのフレームワークです。現在、世界中の企業で広く導入されています。Kafkaの公式サイトでその思想や導入方法について詳しく知ることができます。主要な特徴の1つは、MQTTを含む他のアプリケーションや通信プロトコルへのコネクタが豊富に既存していることです。

MQTT

MQTTは軽量なTCPベースのメッセージングプロトコルで、堅牢性とフットプリントの小ささからIoT通信によく使用されます。OASIS標準のMQTTについて詳しくは公式サイトを参照してください。

こちらで、PLCnext向けにmosquittoをクロスコンパイルするMakers Blog記事が見つかります。mosquittoはEclipseのMQTT実装です。また、PLCnext Storeではすぐに使えるMQTTアプリも提供されています。

要件

• PLCnext上のMQTTクライアント（実装ヒントは前節参照）
• コントローラがPC/VMと接続されていること
• PC/VM上のMQTTブローカー（例：mosquitto）
• PC/VM上のKafkaインスタンス（Kafkaのクイックスタートガイド参照）

セットアップ概要

以下の図は、PLCnext ControlからKafkaへデータを取り込むために実装するセットアップの概要を示しています。ConfluentのMQTT Proxyを彼らのKafka版（2）で使用することも可能ですが、ここではより汎用的なソリューション（1）に焦点を当てます。これは、クライアントが接続してメッセージを発行するMQTTブローカーと、ブローカーのトピックを購読しメッセージを処理してKafkaへ転送するコネクタで構成されます。

コネクタの作成

このチュートリアルでは、GitHubのevokly/kafka-connect-mqttリポジトリ（MITライセンス）をベースにコネクタを作成します。まずリポジトリをダウンロードして展開します。最新リポジトリ版は2016年末のものなので、build.gradleファイルを更新し、古い依存関係を新しいバージョンに置き換えます：

ext { kafkaVersion = '2.6.0' }
...
dependencies {
    testCompile group: 'junit', name: 'junit', version: '4.13'
    compile "org.apache.kafka:connect-api:$kafkaVersion"
    compile 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5'
    compile 'org.bouncycastle:bcprov-jdk15on:1.67'
    compile 'org.bouncycastle:bcpkix-jdk15on:1.67'
    compile 'org.bouncycastle:bcpg-jdk15on:1.67'
    compile 'commons-io:commons-io:2.8.0'
    compile 'org.slf4j:slf4j-api:1.7.30'
    testCompile 'org.slf4j:slf4j-simple:1.7.30'
}

この例ではプレーンなStringメッセージをKafkaに送信します。そのため、デフォルトのメッセージプロセッサである/kafka-connect-mqtt-master/src/main/java/com/evokly/kafka/connect/mqttフォルダ内のJavaクラスDumbProcessor.javaを編集します：

@Override
public SourceRecord[] getRecords(String kafkaTopic) {
    return new SourceRecord[]{
        new SourceRecord(
            null, // sourcePartition
            null, // sourceOffset
            kafkaTopic, // topic
            null, // partition
            null, // keySchema
            mTopic, // key
            null, // valueSchema
            mMessage.toString(), // value
            new Long(123L)
        )
    }; // long timestamp
}

その後、依存関係を含むJava Archive（JAR）をビルドします：./gradlew clean jar。出力されたJARkafka-connect-mqtt-1.1-SNAPSHOT.jar（/kafka-connect-mqtt-master/build/libsにあります）をKafkaのlibsディレクトリにコピーします。

また、org.eclipse.paho.client.mqttv3-1.2.5.jarのコピーもKafkaのlibsディレクトリに配置します。こちらからダウンロード可能です。

さらに、コネクタ用の設定ファイルmqtt.propertiesをKafkaのconfigフォルダに作成します。内容は以下の通りです：

name=mqtt
connector.class=com.evokly.kafka.connect.mqtt.MqttSourceConnector
tasks.max=1
# スキーマなしプレーンStringメッセージ用コンバータ
key.converter = org.apache.kafka.connect.storage.StringConverter
value.converter = org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
kafka.topic=test_in
mqtt.client_id=mqtt-kafka-123
mqtt.clean_session=true
mqtt.connection_timeout=30
mqtt.keep_alive_interval=60
mqtt.server_uris=tcp://172.17.0.1:1883
mqtt.topic=test/#
#if we want to use our own processor class
#message_processor_class=com.evokly.kafka.connect.mqtt.sample.OwnProcessor

ローカルテスト

これでローカルテストが可能です。Kafkaのディレクトリに移動し、ZooKeeperとBrokerインスタンスを起動します：

# ZooKeeper起動
bin/zookeeper-server-start.sh config/zookeeper.properties
# Kafka起動
bin/kafka-server-start.sh config/server.properties
# MQTTブローカー起動（ここではmosquittoのDockerコンテナ）
sudo docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto
# MQTT-Kafkaコネクタ起動
bin/connect-standalone.sh config/connect-standalone.properties config/mqtt.properties
# Kafkaコンソールコンシューマー起動
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_in --from-beginning --property print.value=true --property print.timestamp=true
# MQTTメッセージ発行
mosquitto_pub -h 172.17.0.1 -p 1883 -t test/1 -m test123

メッセージがコンソールコンシューマーに表示されます。