PLCnext Control을 MQTT로 Apache Kafka와 연동하기: 엔터프라이즈급 통합

기술적 배경

Kafka

Apache Kafka는 데이터 수집, 저장, 처리, 재배포를 위한 프레임워크입니다. 현재 전 세계 기업에서 널리 배포되고 있습니다. Kafka 공식 웹사이트에서 개념과 배포 방법에 대한 자세한 정보를 확인할 수 있습니다. 주요 특징 중 하나는 MQTT 같은 다른 애플리케이션 및 통신 프로토콜과의 수많은 기존 커넥터입니다.

MQTT

MQTT는 IoT 통신에 자주 사용되는 경량 TCP 기반 메시징 프로토콜로, 견고성과 작은 풋프린트가 특징입니다. OASIS 표준 MQTT에 대한 자세한 내용은 웹사이트에서 확인할 수 있습니다.

이 메이커스 블로그 기사에서 PLCnext용 모스키토(mosquitto) 크로스 컴파일 방법(Eclipse의 MQTT 구현)을 확인할 수 있습니다. 또는 PLCnext Store에서 바로 사용 가능한 MQTT 앱을 제공합니다.

요구사항

• PLCnext에 MQTT 클라이언트 (구현 힌트는 이전 섹션 참조)
• 컨트롤러가 PC/VM에 연결됨
• PC/VM에 MQTT 브로커 (예: mosquitto)
• PC/VM에 Kafka 인스턴스 (Kafka 퀵스타트 가이드 참조)

설정 개요

다음 그림은 PLCnext 컨트롤에서 Kafka로 데이터를 수집하기 위해 구현할 설정의 개요입니다. Confluent의 MQTT 프록시(2)를 Confluent 버전 Kafka에 사용하는 것도 가능하지만, 여기서는 더 일반적인 솔루션(1)에 집중합니다. 이 솔루션은 클라이언트가 연결하고 메시지를 발행하는 MQTT 브로커, 브로커의 토픽을 구독해 메시지를 처리하고 Kafka로 전달하는 커넥터로 구성됩니다.

커넥터 생성

이 튜토리얼의 커넥터는 GitHub의 evokly/kafka-connect-mqtt 저장소(MIT 라이선스)를 기반으로 합니다. 먼저 저장소를 다운로드하고 압축을 풉니다. 최신 저장소 버전이 2016년 말이라 build.gradle 파일의 오래된 의존성을 새 버전으로 업데이트합니다:

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

이 예제에서는 일반 문자열 메시지를 Kafka로 보냅니다. 따라서 기본 메시지 프로세서인 /kafka-connect-mqtt-master/src/main/java/com/evokly/kafka/connect/mqtt 폴더의 DumbProcessor.java 자바 클래스를 편집합니다:

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

이후 의존성을 포함한 JAR(Java Archive) 파일을 빌드합니다: ./gradlew clean jar. 출력 JAR kafka-connect-mqtt-1.1-SNAPSHOT.jar는 /kafka-connect-mqtt-master/build/libs 폴더에서 찾을 수 있으며, 이를 Kafka의 libs 디렉토리에 복사합니다.

또한 org.eclipse.paho.client.mqttv3-1.2.5.jar 아카이브 복사본이 Kafka의 libs 디렉토리에 필요합니다. 여기서 다운로드할 수 있습니다.

커넥터용 설정 파일 mqtt.properties를 Kafka의 config 폴더에 생성합니다. 파일 내용은 다음과 같습니다:

name=mqtt
connector.class=com.evokly.kafka.connect.mqtt.MqttSourceConnector
tasks.max=1
# 스키마 없는 일반 문자열 메시지용 컨버터
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

로컬 테스트

이제 커넥터를 로컬에서 테스트할 수 있습니다. Kafka 디렉토리로 이동해 ZooKeeper와 브로커 인스턴스를 시작합니다:

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

# Kafka 시작
bin/kafka-server-start.sh config/server.properties

# MQTT 브로커 시작 (여기서는 mosquitto 도커 컨테이너)
sudo docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto

# MQTT-Kafka 커넥터 시작
bin/connect-standalone.sh config/connect-standalone.properties config/mqtt.properties

# Kafka 콘솔 컨슈머 시작
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_in --from-beginning --property print.value=true --property print.timestamp=true

# MQTT 메시지 발행
mosquitto_pub -h 172.17.0.1 -p 1883 -t test/1 -m test123

메시지가 콘솔 컨슈머에 나타납니다.