Verbind PLCnext Control via MQTT met Apache Kafka

Technische achtergrond

Kafka

Apache Kafka is een raamwerk voor gegevensopname, opslag, verwerking en herdistributie. Tegenwoordig wordt het breed ingezet bij bedrijven over de hele wereld. De officiële website van Kafka biedt meer informatie over het idee en hoe het te implementeren. Een van de belangrijkste kenmerken is het enorme aantal reeds bestaande connectoren naar andere applicaties en communicatieprotocollen zoals MQTT.

MQTT

MQTT is een lichtgewicht op TCP gebaseerd berichtenprotocol, dat vanwege zijn robuustheid en kleine footprint vaak wordt gebruikt voor IoT-communicatie. Details over de OASIS-standaard MQTT zijn te vinden op haar website.

Hier vindt u een Makers Blog-artikel over het cross-compileren van mosquitto voor PLCnext, een MQTT-implementatie van Eclipse. Als alternatief biedt de PLCnext Store kant-en-klare MQTT-apps.

Vereisten

• MQTT-client op de PLCnext (zie de vorige sectie voor implementatietips)
• de controller is aangesloten op een pc/VM
• MQTT-makelaar op de pc/VM (bijv. Mosquitto)
• Kafka-instantie op de pc/VM (zie de snelstartgids van Kafka)

Instellen

De volgende afbeelding toont een overzicht van de setup die we gaan implementeren om gegevens van de PLCnext-besturing in Kafka op te nemen. Hoewel het mogelijk is om de MQTT Proxy van Confluent te gebruiken voor hun versie van Kafka (2), zullen we ons concentreren op de meer generieke oplossing (1). Het bestaat uit een MQTT-broker waar de klant verbinding mee maakt en berichten publiceert en een connector die zich abonneert op een onderwerp bij de broker, de berichten verwerkt en doorstuurt naar Kafka.

De connector maken

In deze tutorial is onze connector gebaseerd op de evokly/kafka-connect-mqtt-repository van GitHub, gelicentieerd onder de MIT-licentie (gedetailleerde licentie-informatie). Eerst downloaden en extraheren we de repository. Aangezien de laatste repository-versie van eind 2016 is, updaten we de build.gradle bestand, 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 gewone String-berichten naar Kafka. Daarom moeten we de Java-klasse DumbProcessor.java . aanpassen in de map /kafka-connect-mqtt-master/src/main/java/com/evokly/kafka/connect/mqtt , wat de standaard berichtenverwerker 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 File (JAR) dat de afhankelijkheden bevat:./gradlew clean jar . We kopiëren de output JAR kafka-connect-mqtt-1.1-SNAPSHOT.jar die te vinden was in de map /kafka-connect-mqtt-master/build/libs naar de libs directory van Kafka.

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

Verder moeten we een configuratiebestand maken voor de connector mqtt.properties in Kafka's config map. Het bestand heeft de volgende inhoud:

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

Lokale test

Nu kunnen we onze connector lokaal testen. Ga naar de directory van Kafka 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 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

Het bericht verschijnt in de consoleconsument.