Conecte PLCnext Control a través de MQTT a Apache Kafka

Antecedentes técnicos

Kafka

Apache Kafka es un marco para la ingesta, el almacenamiento, el procesamiento y la redistribución de datos. Hoy en día, está ampliamente desplegado en empresas de todo el mundo. El sitio web oficial de Kafka ofrece más información sobre su idea y cómo implementarla. Una de sus características clave es la gran cantidad de conectores ya existentes a otras aplicaciones y protocolos de comunicación como MQTT.

MQTT

MQTT es un protocolo de mensajería ligero basado en TCP, que a menudo se usa para la comunicación IoT debido a su robustez y tamaño reducido. Los detalles sobre el estándar MQTT de OASIS se pueden encontrar en su sitio web.

Aquí puede encontrar un artículo del Makers Blog sobre cómo realizar una compilación cruzada de mosquitto para PLCnext, una implementación de MQTT de Eclipse. Como alternativa, la tienda PLCnext ofrece aplicaciones MQTT listas.

Requisitos

• Cliente MQTT en PLCnext (consulte la sección anterior para obtener sugerencias de implementación)
• el controlador está conectado a una PC/VM
• Broker MQTT en la PC/VM (por ejemplo, mosquitto)
• Instancia de Kafka en la PC/VM (consulte la guía de inicio rápido de Kafka)

Configuración

La siguiente imagen muestra una descripción general de la configuración que vamos a implementar para ingerir datos desde el control PLCnext a Kafka. Si bien es posible utilizar el proxy MQTT de Confluent para su versión de Kafka (2), nos centraremos en la solución más genérica (1). Comprende un bróker MQTT donde el cliente se conecta y publica mensajes y un conector que se suscribe a un tema en el bróker, procesa los mensajes y los reenvía a Kafka.

Creación del conector

En este tutorial nuestro conector se basa en el repositorio evokly/kafka-connect-mqtt de GitHub, licenciado bajo la Licencia MIT (información detallada de la licencia). Primero, descargamos y extraemos el repositorio. Dado que la última versión del repositorio es de finales de 2016, actualizamos el build.gradle archivo, reemplazando dependencias antiguas con sus nuevas versiones:

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

En este ejemplo, enviaremos mensajes String sin formato a Kafka. Por lo tanto tenemos que editar la clase Java DumbProcessor.java en la carpeta /kafka-connect-mqtt-master/src/main/java/com/evokly/kafka/connect/mqtt , que es el procesador de mensajes predeterminado:

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

A partir de entonces, construimos un Archivo de Archivo Java (JAR) que contiene las dependencias:./gradlew clean jar . Copiamos el JAR de salida kafka-connect-mqtt-1.1-SNAPSHOT.jar que podría encontrarse en la carpeta /kafka-connect-mqtt-master/build/libs al libs directorio de Kafka.

También necesitamos una copia del archivo org.eclipse.paho.client.mqttv3-1.2.5.jar en el directorio libs de Kafka. Podemos descargarlo aquí.

Además, tenemos que crear un archivo de configuración para el conector mqtt.properties en los config de Kafka carpeta. El archivo tiene el siguiente contenido:

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

Prueba local

Ahora podemos probar nuestro conector localmente. Vaya al directorio de Kafka e inicie una instancia de ZooKeeper and Broker:

# 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

El mensaje aparece en el consumidor de la consola.