Verbinden Sie PLCnext Control über MQTT mit Apache Kafka

Technischer Hintergrund

Kafka

Apache Kafka ist ein Framework für die Aufnahme, Speicherung, Verarbeitung und Weiterverteilung von Daten. Heutzutage wird es in Unternehmen auf der ganzen Welt eingesetzt. Die offizielle Website von Kafka bietet weitere Informationen über seine Idee und wie man sie einsetzt. Eines seiner Hauptmerkmale ist die große Anzahl bereits vorhandener Konnektoren zu anderen Anwendungen und Kommunikationsprotokollen wie MQTT.

MQTT

MQTT ist ein leichtgewichtiges TCP-basiertes Messaging-Protokoll, das aufgrund seiner Robustheit und seines geringen Platzbedarfs häufig für die IoT-Kommunikation verwendet wird. Details zum OASIS-Standard MQTT finden Sie auf dessen Website.

Hier finden Sie einen Makers-Blog-Artikel über die Cross-Kompilierung von Mosquitto für PLCnext, eine MQTT-Implementierung von Eclipse. Alternativ bietet der PLCnext Store fertige MQTT-Apps an.

Anforderungen

• MQTT-Client auf der PLCnext (Implementierungshinweise siehe vorheriger Abschnitt)
• der Controller ist mit einem PC/einer VM verbunden
• MQTT-Broker auf dem PC/VM (z. B. Mosquitto)
• Kafka-Instanz auf dem PC/VM (siehe Schnellstartanleitung von Kafka)

Einrichtung

Das folgende Bild zeigt einen Überblick über das Setup, das wir implementieren werden, um Daten von der PLCnext-Steuerung in Kafka zu importieren. Obwohl es möglich ist, den MQTT-Proxy von Confluent für ihre Version von Kafka (2) zu verwenden, konzentrieren wir uns auf die allgemeinere Lösung (1). Es besteht aus einem MQTT-Broker, mit dem sich der Client verbindet und Nachrichten veröffentlicht, und einem Konnektor, der ein Thema beim Broker abonniert, die Nachrichten verarbeitet und an Kafka weiterleitet.

Konnektor erstellen

In diesem Tutorial basiert unser Konnektor auf dem evokly/kafka-connect-mqtt-Repository von GitHub, lizenziert unter der MIT-Lizenz (detaillierte Lizenzinformationen). Zuerst laden wir das Repository herunter und extrahieren es. Da die neueste Repository-Version von Ende 2016 ist, aktualisieren wir den build.gradle Datei, indem Sie alte Abhängigkeiten durch ihre neuen Versionen ersetzen:

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'
}

In diesem Beispiel senden wir einfache String-Nachrichten an Kafka. Dazu müssen wir die Java-Klasse DumbProcessor.java bearbeiten im Ordner /kafka-connect-mqtt-master/src/main/java/com/evokly/kafka/connect/mqtt , das ist der standardmäßige Nachrichtenprozessor:

@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
}

Danach erstellen wir eine Java-Archivdatei (JAR), die die Abhängigkeiten enthält:./gradlew clean jar . Wir kopieren die Ausgabe JAR kafka-connect-mqtt-1.1-SNAPSHOT.jar die sich im Ordner /kafka-connect-mqtt-master/build/libs befinden zum libs Verzeichnis von Kafka.

Außerdem benötigen wir eine Kopie des Archivs org.eclipse.paho.client.mqttv3-1.2.5.jar im libs-Verzeichnis von Kafka. Wir können es hier herunterladen.

Außerdem müssen wir eine Konfigurationsdatei für den Connector mqtt.properties erstellen in Kafkas config Mappe. Die Datei hat folgenden Inhalt:

name=mqtt
connector.class=com.evokly.kafka.connect.mqtt.MqttSourceConnector
tasks.max=1

# converters for plain String messages without schemas
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                     # Kafka destination topic for the MQTT messages
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  # address of the MQTT broker
mqtt.topic=test/#                       # MQTT topic where the messages should be collected

#if we want to use our own processor class
#message_processor_class=com.evokly.kafka.connect.mqtt.sample.OwnProcessor

Lokaler Test

Jetzt können wir unseren Connector lokal testen. Gehen Sie in das Kafka-Verzeichnis und starten Sie eine ZooKeeper- und Broker-Instanz:

# start ZooKeeper
bin/zookeeper-server-start.sh config/zookeeper.properties

# start Kafka:
bin/kafka-server-start.sh config/server.properties

# start an MQTT-Broker (here a mosquitto docker container)
sudo docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto 

# start the MQTT-Kafka connector
bin/connect-standalone.sh config/connect-standalone.properties config/mqtt.properties

# start a Kafka console consumer
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_in --from-beginning --property print.value=true --property print.timestamp=true

# publish an MQTT message
mosquitto_pub -h 172.17.0.1 -p 1883 -t test/1 -m test123

Die Nachricht wird im Konsolenkonsumenten angezeigt.