{"id":1076,"date":"2021-05-02T14:20:07","date_gmt":"2021-05-02T12:20:07","guid":{"rendered":"https:\/\/www.rocworks.at\/wordpress\/?p=1076"},"modified":"2021-05-03T11:07:49","modified_gmt":"2021-05-03T09:07:49","slug":"how-to-log-opc-ua-tag-values-to-apache-kafka","status":"publish","type":"post","link":"https:\/\/www.rocworks.at\/wordpress\/?p=1076","title":{"rendered":"How to log OPC UA tag values to Apache Kafka…"},"content":{"rendered":"\n

In this article we use the Frankenstein Automation Gateway<\/a> to subscribe to one public available OPC UA server (milo.digitalpetri.com) and log tag values to Apache Kafka<\/a>. Additionally we show how you can create a Stream in Apache Kafka based on the OPC UA values coming from the milo OPC UA server and query those stream with KSQL<\/a>. <\/p>\n\n\n\n

Setup Apache Kafka<\/strong><\/p>\n\n\n\n

We have used the all-in-one<\/a> Docker compose file from confluent to quickly setup Apache Kafka and KSQL. Be sure that you set your resolvable hostname or IP address of your server in the docker-compose.yml file. Otherwise Kafka clients cannot connect to the broker.<\/p>\n\n\n\n

KAFKA_ADVERTISED_LISTENERS: PLAINTEXT:\/\/broker:29092,PLAINTEXT_HOST:\/\/192.168.1.18<\/strong>:9092<\/p>\n\n\n\n

Setup Frankenstein<\/strong><\/p>\n\n\n\n

Install Java 11 (for example Amazon Corretto<\/a>) and Gradle<\/a> for Frankenstein. Unzip Gradle to a folder and set your PATH variable to point to the bin directory of Gradle.<\/p>\n\n\n\n

Then clone the source of Frankenstein and compile it with Gradle:<\/p>\n\n\n\n

git clone https:\/\/github.com\/vogler75\/automation-gateway.git\ncd automation-gateway\/source\/app\ngradle build<\/code><\/pre>\n\n\n\n
There is a example config-milo-kafka.yaml file in the automation-gateway\/source\/app directory which you can use by setting the environment variable GATEWAY_CONFIG. <\/p>\n\n\n\n
export GATEWAY_CONFIG=config-milo-kafka.yaml\n<\/code><\/pre>\n\n\n\n
In this config file we use a public Eclipse Milo OPC UA server. The Id of this connection is “milo<\/strong>“.<\/p>\n\n\n\n
OpcUaClient:\n  - Id: \"milo\"\n    Enabled: true\n    LogLevel: INFO\n    EndpointUrl: \"opc.tcp:\/\/milo.digitalpetri.com:62541\/milo\"\n    UpdateEndpointUrl: false\n    SecurityPolicyUri: http:\/\/opcfoundation.org\/UA\/SecurityPolicy#None\n    UsernameProvider:\n      Username: user1\n      Password: password<\/code><\/pre>\n\n\n\n
Here is the configuration of the Kafka Logger where you can configure what OPC UA tags should be published to Kafka. In that case we use a OPC UA Browse Path and a wildcard to use all variables below one node. <\/p>\n\n\n\n
Database:\n  Logger:\n    - Id: kafka1\n      Type: Kafka\n      Enabled: true\n      Servers: server2:9092\n      WriteParameters:\n        QueueSize: 20000\n        BlockSize: 10000\n      Logging:\n        - Topic: opc\/milo\/path\/Objects\/Dynamic\/+<\/code><\/pre>\n\n\n\n
Start Frankenstein<\/strong><\/p>\n\n\n\n
export GATEWAY_CONFIG=config-milo-kafka.yaml\ngradle run<\/code><\/pre>\n\n\n\n
Create a Stream in KSQL<\/strong><\/p>\n\n\n\n
Start a CLI session to KSQL on the host where the Kafka containers run: <\/p>\n\n\n\n
docker exec -ti ksqldb-cli ksql http:\/\/ksqldb-server:8088<\/code><\/pre>\n\n\n\n
Create a stream for the Kafka “milo” topic<\/p>\n\n\n\n
CREATE STREAM milo(\n  browsePath VARCHAR KEY, \n  sourceTime VARCHAR, \n  value DOUBLE, \n  statusCode VARCHAR\n) WITH (\n  KEY_FORMAT='KAFKA',\n  KAFKA_TOPIC='milo', \n  VALUE_FORMAT='JSON',\n  TIMESTAMP='sourceTime',TIMESTAMP_FORMAT='yyyy-MM-dd''T''HH:mm:ss.nX'\n);<\/code><\/pre>\n\n\n\n
Then you can execute a KSQL query to get the stream of values from the OPC UA server:<\/p>\n\n\n\n
ksql> select browsepath, sourcetime, value from milo emit changes;\n+---------------------------------------+---------------------------------------+---------------------------------------+\n|BROWSEPATH                             |SOURCETIME                             |VALUE                                  |\n+---------------------------------------+---------------------------------------+---------------------------------------+\n|Objects\/Dynamic\/RandomInt32            |2021-05-02T11:29:04.405465Z            |1489592303                             |\n|Objects\/Dynamic\/RandomInt64            |2021-05-02T11:29:04.405322Z            |-6.3980451035323023E+18                |\n|Objects\/Dynamic\/RandomFloat            |2021-05-02T11:29:04.405350Z            |0.7255345                              |\n|Objects\/Dynamic\/RandomDouble           |2021-05-02T11:29:04.405315Z            |0.23769088795602633                    |\n<\/code><\/pre>\n\n\n\n
<\/p>\n","protected":false},"excerpt":{"rendered":"
In this article we use the Frankenstein Automation Gateway to subscribe to one public available OPC UA server (milo.digitalpetri.com) and log tag values to Apache Kafka. Additionally we show how you can create a Stream in Apache Kafka based on … Continue reading →<\/span><\/a><\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1,39],"tags":[37,36,35,38],"class_list":["post-1076","post","type-post","status-publish","format-standard","hentry","category-allgemein","category-frankenstein","tag-kafka","tag-ksq","tag-opcua","tag-streaming"],"_links":{"self":[{"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/1076","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=1076"}],"version-history":[{"count":16,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/1076\/revisions"}],"predecessor-version":[{"id":1094,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/1076\/revisions\/1094"}],"wp:attachment":[{"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=1076"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=1076"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=1076"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}