Conecte PLCnext Control ao Apache Kafka via MQTT – Integração Enterprise

Fundamento Técnico

Kafka

O Apache Kafka é um framework para ingestão, armazenamento, processamento e redistribuição de dados. Atualmente, é amplamente implantado em empresas em todo o mundo. O site oficial do Kafka oferece mais informações sobre sua ideia e como implantá-lo. Uma de suas principais características é o grande número de conectores já existentes para outras aplicações e protocolos de comunicação como MQTT.

MQTT

MQTT é um protocolo de mensageria leve baseado em TCP, frequentemente usado para comunicação IoT devido à sua robustez e pequeno footprint. Detalhes sobre o padrão OASIS MQTT podem ser encontrados em seu site.

Aqui você pode encontrar um artigo do Makers Blog sobre como fazer cross-compile do mosquitto para PLCnext, uma implementação MQTT da Eclipse. Alternativamente, a PLCnext Store oferece apps MQTT prontos.

Requisitos

• Cliente MQTT no PLCnext (veja a seção anterior para dicas de implementação)
• o controlador está conectado a um PC/VM
• Broker MQTT no PC/VM (ex: mosquitto)
• Instância Kafka no PC/VM (veja guia de início rápido do Kafka)

Configuração

A imagem a seguir mostra uma visão geral da configuração que vamos implementar para ingerir dados do controle PLCnext para o Kafka. Embora seja possível usar o MQTT Proxy da Confluent para sua versão do Kafka (2), focaremos na solução mais genérica (1). Ela compreende um broker MQTT onde o cliente conecta e publica mensagens e um conector que se inscreve em um tópico no broker, processa as mensagens e as encaminha para o Kafka.

Criando o Conector

Neste tutorial, nosso conector baseia-se no repositório evokly/kafka-connect-mqtt do GitHub, licenciado sob MIT License. Primeiro, baixamos e extraímos o repositório. Como a versão mais recente do repositório é do final de 2016, atualizamos o arquivo build.gradle, substituindo dependências antigas por suas novas versões:

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

Neste exemplo, enviaremos mensagens String simples para o Kafka. Portanto, temos que editar a classe Java DumbProcessor.java na pasta /kafka-connect-mqtt-master/src/main/java/com/evokly/kafka/connect/mqtt, que é o processador de mensagens padrão:

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

Em seguida, construímos um Java Archive File (JAR) que contém as dependências: ./gradlew clean jar. Copiamos o JAR de saída kafka-connect-mqtt-1.1-SNAPSHOT.jar que pode ser encontrado na pasta /kafka-connect-mqtt-master/build/libs para o diretório libs do Kafka.

Também precisamos de uma cópia do arquivo org.eclipse.paho.client.mqttv3-1.2.5.jar no diretório libs do Kafka. Podemos baixá-lo aqui.

Além disso, temos que criar um arquivo de configuração para o conector mqtt.properties na pasta config do Kafka. O arquivo tem o seguinte conteúdo:

name=mqtt
connector.class=com.evokly.kafka.connect.mqtt.MqttSourceConnector
tasks.max=1
# conversores para mensagens String simples sem 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 # tópico Kafka de destino para mensagens 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 # endereço do broker MQTT
mqtt.topic=test/# # tópico MQTT onde as mensagens devem ser coletadas
#if we want to use our own processor class
#message_processor_class=com.evokly.kafka.connect.mqtt.sample.OwnProcessor

Teste Local

Agora podemos testar nosso conector localmente. Vá para o diretório do Kafka e inicie uma instância ZooKeeper e Broker:

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

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

# inicia um MQTT-Broker (aqui um container docker mosquitto)
sudo docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto

# inicia o conector MQTT-Kafka
bin/connect-standalone.sh config/connect-standalone.properties config/mqtt.properties

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

# publica uma mensagem MQTT
mosquitto_pub -h 172.17.0.1 -p 1883 -t test/1 -m test123

A mensagem aparece no consumidor de console.