Conecta PLCnext Control a Apache Kafka usando MQTT: Integración empresarial

Contexto técnico

Kafka

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

MQTT

MQTT es un protocolo de mensajería ligero basado en TCP, usado a menudo para comunicación IoT debido a su robustez y pequeña huella. Detalles sobre el estándar OASIS MQTT se pueden encontrar en su sitio web.

Aquí puedes encontrar un artículo del Makers Blog sobre cómo compilar mosquitto para PLCnext, una implementación MQTT de Eclipse. Alternativamente, la PLCnext Store ofrece apps MQTT listas para usar.

Requisitos

• Cliente MQTT en el PLCnext (ver sección anterior para pistas de implementación)
• El controlador conectado a un PC/VM
• Broker MQTT en el PC/VM (ej. mosquitto)
• Instancia Kafka en el PC/VM (ver guía de inicio rápido de Kafka)

Configuración

La siguiente imagen muestra una visión general de la configuración que vamos a implementar para ingerir datos del control PLCnext a Kafka. Aunque es posible usar el MQTT Proxy de Confluent para su versión de Kafka (2) nos centraremos en la solución más genérica (1). Comprende un broker MQTT donde el cliente se conecta y publica mensajes y un conector que se suscribe a un tema en el broker, 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 archivo build.gradle, reemplazando las 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 simples a Kafka. Por 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 por defecto:

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

Después, construimos un archivo Java Archive (JAR) que contiene las dependencias: ./gradlew clean jar. Copiamos el JAR de salida kafka-connect-mqtt-1.1-SNAPSHOT.jar que se encuentra en la carpeta /kafka-connect-mqtt-master/build/libs al directorio libs 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 la carpeta config de Kafka. El archivo tiene el siguiente contenido:

name=mqtt
connector.class=com.evokly.kafka.connect.mqtt.MqttSourceConnector
tasks.max=1
# conversores para mensajes String simples sin esquemas
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 # tema destino Kafka para mensajes MQTT
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 # dirección del broker MQTT
mqtt.topic=test/# # tema MQTT donde recopilar los mensajes
#if queremos usar nuestra propia clase procesadora
#message_processor_class=com.evokly.kafka.connect.mqtt.sample.OwnProcessor

Prueba local

Ahora podemos probar nuestro conector localmente. Ve al directorio de Kafka e inicia una instancia de ZooKeeper y Broker:

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

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

# iniciar un Broker MQTT (aquí un contenedor docker mosquitto)
sudo docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto

# iniciar el conector MQTT-Kafka
bin/connect-standalone.sh config/connect-standalone.properties config/mqtt.properties

# iniciar un consumidor de consola Kafka
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_in --from-beginning --property print.value=true --property print.timestamp=true

# publicar un mensaje MQTT
mosquitto_pub -h 172.17.0.1 -p 1883 -t test/1 -m test123

El mensaje aparece en el consumidor de consola.