PLCnext Control per MQTT nahtlos an Apache Kafka anbinden – Enterprise-Integration

Technischer Hintergrund

Kafka

Apache Kafka ist ein Framework für Datenerfassung, -speicherung, -verarbeitung und -umverteilung. Weltweit weit verbreitet. Eine Schlüsselfunktion: die große Anzahl bestehender Konnektoren zu anderen Anwendungen und Protokollen wie MQTT. Mehr auf der offiziellen Kafka-Website.

MQTT

MQTT ist ein leichtgewichtiges, TCP-basiertes Messaging-Protokoll, häufig für IoT-Kommunikation genutzt aufgrund von Robustheit und kleinem Footprint. Details zum OASIS-Standard auf der MQTT-Website.

Ein Makers-Blog-Artikel zeigt, wie man Mosquitto für PLCnext cross-kompiliert. Alternativ bietet der PLCnext Store fertige MQTT-Apps.

Voraussetzungen

• MQTT-Client auf dem PLCnext (siehe vorheriger Abschnitt)
• Controller mit PC/VM verbunden
• MQTT-Broker auf PC/VM (z. B. Mosquitto)
• Kafka-Instanz auf PC/VM (Kafka Quickstart Guide)

Aufbau

Das folgende Bild zeigt den Aufbau zur Datenerfassung vom PLCnext Control zu Kafka. Während Confluent's MQTT Proxy für ihre Kafka-Version nutzbar ist (2), fokussieren wir uns auf die generischere Lösung (1): Ein MQTT-Broker, zu dem der Client verbindet und Nachrichten veröffentlicht, und ein Konnektor, der ein Topic beim Broker abonniert, Nachrichten verarbeitet und an Kafka weiterleitet.

Konnektor erstellen

Basiert auf dem evokly/kafka-connect-mqtt-Repository (GitHub, MIT-Lizenz). Repository herunterladen und entpacken. Da die neueste Version von Ende 2016 stammt, build.gradle aktualisieren – alte Dependencies durch neue 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 Plain-String-Nachrichten an Kafka. Daher die Java-Klasse DumbProcessor.java im Ordner /kafka-connect-mqtt-master/src/main/java/com/evokly/kafka/connect/mqtt bearbeiten (Standard-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)
    )
  };
}

Danach Java Archive (JAR) mit Dependencies bauen: ./gradlew clean jar. Output-JAR kafka-connect-mqtt-1.1-SNAPSHOT.jar aus /kafka-connect-mqtt-master/build/libs in das libs-Verzeichnis von Kafka kopieren.

Außerdem org.eclipse.paho.client.mqttv3-1.2.5.jar in das libs-Verzeichnis von Kafka kopieren (Download hier).

Konfigurationsdatei mqtt.properties im config-Ordner von Kafka erstellen:

name=mqtt
connector.class=com.evokly.kafka.connect.mqtt.MqttSourceConnector
tasks.max=1
# Konverter für Plain-String-Nachrichten ohne 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
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/#
# Eigener Prozessor (optional)
# message_processor_class=com.evokly.kafka.connect.mqtt.sample.OwnProcessor

Lokaler Test

In Kafka-Verzeichnis wechseln und ZooKeeper sowie Broker starten:

# ZooKeeper starten
bin/zookeeper-server-start.sh config/zookeeper.properties
# Kafka starten
bin/kafka-server-start.sh config/server.properties
# MQTT-Broker starten (Mosquitto Docker Container)
sudo docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto
# MQTT-Kafka-Konnektor starten
bin/connect-standalone.sh config/connect-standalone.properties config/mqtt.properties
# Kafka Console Consumer starten
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_in --from-beginning --property print.value=true --property print.timestamp=true
# MQTT-Nachricht veröffentlichen
mosquitto_pub -h 172.17.0.1 -p 1883 -t test/1 -m test123

Die Nachricht erscheint im Console Consumer.