PLCnext Control naadloos koppelen aan Apache Kafka via MQTT – Enterprise-integratie

Technische achtergrond

Kafka

Apache Kafka is een framework voor datainname, opslag, verwerking en herdistributie. Tegenwoordig wordt het wereldwijd bij bedrijven geïmplementeerd. De officiële website van Kafka biedt meer informatie over het concept en de implementatie. Een van de belangrijkste kenmerken is het grote aantal al bestaande connectors naar andere applicaties en communicatieprotocollen zoals MQTT.

MQTT

MQTT is een lichtgewicht, op TCP gebaseerd messaging-protocol, vaak gebruikt voor IoT-communicatie vanwege zijn robuustheid en kleine voetafdruk. Details over de OASIS-standaard MQTT vind je op de website.

Hier vind je een Makers Blog-artikel over het cross-compileren van mosquitto voor PLCnext, een MQTT-implementatie van Eclipse. Alternatief biedt de PLCnext Store kant-en-klare MQTT-apps.

Vereisten

• MQTT-client op de PLCnext (zie vorige sectie voor implementatiehints)
• De controller is verbonden met een PC/VM
• MQTT-broker op de PC/VM (bijv. mosquitto)
• Kafka-instantie op de PC/VM (zie Kafka's quickstart guide)

Opzet

De volgende afbeelding toont een overzicht van de opzet die we gaan implementeren om data van de PLCnext-control naar Kafka te transporteren. Hoewel het mogelijk is om Confluent's MQTT Proxy te gebruiken voor hun versie van Kafka (2), richten we ons op de meer generieke oplossing (1). Deze omvat een MQTT-broker waar de client verbinding mee maakt en berichten publiceert, en een connector die zich abonneert op een topic bij de broker, de berichten verwerkt en ze naar Kafka doorstuurt.

De connector maken

In deze tutorial is onze connector gebaseerd op de evokly/kafka-connect-mqtt repository op GitHub, gelicenseerd onder de MIT-licentie. Eerst downloaden en uitpakken we de repository. Omdat de laatste repositoryversie uit eind 2016 dateert, werken we het build.gradle-bestand bij door oude afhankelijkheden te vervangen door hun nieuwe versies:

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 dit voorbeeld sturen we platte String-berichten naar Kafka. Daarom moeten we de Java-klasse DumbProcessor.java in de map /kafka-connect-mqtt-master/src/main/java/com/evokly/kafka/connect/mqtt aanpassen, wat de standaard berichtverwerker is:

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

Daarna bouwen we een Java Archive (JAR) met de afhankelijkheden: ./gradlew clean jar. We kopiëren de uitvoer-JAR kafka-connect-mqtt-1.1-SNAPSHOT.jar (te vinden in /kafka-connect-mqtt-master/build/libs) naar de libs-directory van Kafka.

We hebben ook een kopie nodig van org.eclipse.paho.client.mqttv3-1.2.5.jar in de libs-directory van Kafka. Die kunnen we hier downloaden.

Verder moeten we een configuratiebestand voor de connector aanmaken: mqtt.properties in Kafka's config-map. De inhoud:

name=mqtt
connector.class=com.evokly.kafka.connect.mqtt.MqttSourceConnector
tasks.max=1
# converters voor platte String-berichten zonder 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/#
#if we want to use our own processor class
#message_processor_class=com.evokly.kafka.connect.mqtt.sample.OwnProcessor

Lokale test

Nu kunnen we onze connector lokaal testen. Ga naar Kafka's directory en start een ZooKeeper- en Broker-instantie:

# start ZooKeeper
bin/zookeeper-server-start.sh config/zookeeper.properties
# start Kafka:
bin/kafka-server-start.sh config/server.properties
# start een MQTT-broker (hier een mosquitto docker container)
sudo docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto
# start de MQTT-Kafka connector
bin/connect-standalone.sh config/connect-standalone.properties config/mqtt.properties
# start een 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
# publiceer een MQTT-bericht
mosquitto_pub -h 172.17.0.1 -p 1883 -t test/1 -m test123

Het bericht verschijnt in de console consumer.