Azure Event Hubs Integration

How to integrate Azure Event Hubs with Connectware, including connecting to the Kafka endpoint of Event Hubs, streaming shop floor data into an event hub, and consuming events from an event hub.

This guide describes how to integrate Azure Event Hubs with Connectware. You configure a service commissioning file that streams shop floor data into an event hub through the Apache Kafka endpoint of Event Hubs and consumes events from an event hub in return. A complete example file is available at the end of this guide.

Objectives

• Establishing a connection between Connectware and the Kafka endpoint of Azure Event Hubs.

• Streaming shop floor data from MQTT topics into an event hub.

• Consuming events from an event hub and publishing them to an MQTT topic.

• Verifying the data flow in both directions.

Prerequisites

To follow this guide, you will need the following:

• A running instance of Cybus Connectware.

• An Azure Event Hubs namespace in the standard, premium, or dedicated tier, with at least one event hub. The basic tier does not support the Kafka endpoint. For more information, see Azure Event Hubs for Apache Kafka in the Azure documentation.

• The connection string of your Event Hubs namespace, available under Shared access policies in the Azure portal. For more information, see Get an Event Hubs connection string in the Azure documentation.

• Access to the Admin UI with sufficient user permissions.

• Basic knowledge of MQTT and the Connectware services concept (for example, service commissioning files, connections, and endpoints).

Connectware and Azure Event Hubs Integration

Azure Event Hubs exposes an Apache Kafka-compatible endpoint on every namespace, reachable at <namespace>.servicebus.windows.net on port 9093. Connectware communicates with this endpoint through the Kafka connector, so no Azure-specific client or SDK is required.

The Kafka concepts map directly to Event Hubs concepts:

Apache Kafka Concept	Event Hubs Concept
Cluster	Namespace
Topic	An event hub
Consumer group	Consumer group

Authentication uses SASL PLAIN over TLS with a shared access signature (SAS): the username is the literal string $ConnectionString, and the password is the connection string of your namespace. When you configure the sasl property, the Kafka connector automatically connects over TLS, which Event Hubs requires for all traffic.

A Kafka connection in Connectware is either a producer or a consumer, controlled by the clientType property. This guide therefore uses two connection resources: one that produces shop floor data into a telemetry event hub, and one that consumes from a commands event hub.

The MQTT topics in this guide follow an ISA-95-style equipment hierarchy (<enterprise>/<site>/<area>/<line>/<cell>). The mapping subscribes with wildcards across all levels, so any machine in the hierarchy is picked up without changing the integration.

This guide covers streaming data into Event Hubs as the entry point to Azure analytics services such as Stream Analytics, Fabric, or Databricks. If you want to connect Connectware as a device to Azure IoT Hub instead, see Azure IoT Hub Integration.

Azure Event Hubs Connection Properties

The connection to Event Hubs requires values from your Azure subscription. We add them as parameters to the service commissioning file, so you can set them when you install the service.

Do not worry about copying the service commissioning file snippets together into one, the complete example file is available at the end of this guide.

• eventHubsNamespace: The name of your Event Hubs namespace, without the domain suffix. For example, example-namespace if your namespace is reachable at example-namespace.servicebus.windows.net.

• connectionString: The connection string of your namespace, starting with Endpoint=sb://. A policy with the Send claim is sufficient for producing, the Listen claim is required for consuming.

• telemetryEventHub: The name of the event hub that receives the shop floor data.

• commandsEventHub: The name of the event hub that Connectware consumes from.

• consumerGroupId: The Kafka consumer group ID used when consuming.

• topicRoot: The root of the MQTT topic hierarchy. Defaults to enterprise.

parameters:
  eventHubsNamespace:
    description: Name of your Event Hubs namespace, without the domain suffix
    type: string
    default: example-namespace

  connectionString:
    description: Connection string of your Event Hubs namespace
    type: string

  telemetryEventHub:
    description: Name of the event hub that receives shop floor data
    type: string
    default: factory-telemetry

  commandsEventHub:
    description: Name of the event hub that Connectware consumes from
    type: string
    default: factory-commands

  consumerGroupId:
    description: Kafka consumer group ID used when consuming from Event Hubs
    type: string
    default: connectware

  topicRoot:
    description: Root of the MQTT topic hierarchy
    type: string
    default: enterprise

Kafka Producer Connection

To produce data into Event Hubs, we set up a Cybus::Connection resource that uses the Kafka connector. The broker address is the Kafka endpoint of the namespace on port 9093. The sasl property carries the SAS authentication and enables TLS. For all available properties, see the Kafka connection properties.

resources:
  eventHubsProducer:
    type: Cybus::Connection
    properties:
      protocol: Kafka
      connection:
        brokers:
          - !sub '${eventHubsNamespace}.servicebus.windows.net:9093'
        sasl:
          mechanism: plain
          username: '$ConnectionString'
          password: !ref connectionString

The username is not a placeholder. Event Hubs expects the literal string $ConnectionString when authenticating with a connection string.

Streaming Shop Floor Data into an Event Hub

We define a write endpoint whose topic is the name of the event hub, and a mapping that feeds it from the MQTT topic hierarchy. The Kafka connector expects a JSON message with the payload in the value property, and each Kafka message carries its body as a string (see Message assembly). The transform rule serializes the incoming payload with the JSONata $string() function and wraps it accordingly, so machines can publish their data without knowing about this convention.

telemetryEndpoint:
  type: Cybus::Endpoint
  properties:
    protocol: Kafka
    connection: !ref eventHubsProducer
    write:
      topic: !ref telemetryEventHub

telemetryMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          topic: !sub '${topicRoot}/+site/+area/+line/+cell/telemetry'
        publish:
          endpoint: !ref telemetryEndpoint
        rules:
          - transform:
              expression: '{ "value": { "value": $string($) } }'

Any message published to a matching topic, for example enterprise/hamburg/assembly/line-1/press-01/telemetry, now becomes one event in the telemetry event hub:

{
  "temperature": 42.1,
  "pressure": 1.8,
  "serial": "SN-0001"
}

The result of every write operation is published to the /res topic of the endpoint. On success, the result contains errorCode: 0 and the offset of the produced event.

Leave the compression property of the endpoint at its default None on the standard tier. Event Hubs supports GZIP-compressed Kafka traffic only in the premium and dedicated tiers.

Consuming Events from an Event Hub

For the opposite direction, we set up a second connection with clientType: consumer, a subscribe endpoint for the commands event hub, and a mapping that publishes every consumed event to an MQTT topic, where any other Connectware service can pick it up, for example to write a command to a PLC.

Kafka consumer groups on Event Hubs are created automatically when the consumer connects, so you do not need to create the consumer group in the Azure portal. A Kafka consumer group spans the whole namespace and is separate from the Event Hubs consumer groups shown in the portal. For more information, see the Event Hubs for Apache Kafka FAQ in the Azure documentation.

eventHubsConsumer:
  type: Cybus::Connection
  properties:
    protocol: Kafka
    connection:
      clientType: consumer
      groupId: !ref consumerGroupId
      brokers:
        - !sub '${eventHubsNamespace}.servicebus.windows.net:9093'
      sasl:
        mechanism: plain
        username: '$ConnectionString'
        password: !ref connectionString

commandsEndpoint:
  type: Cybus::Endpoint
  properties:
    protocol: Kafka
    connection: !ref eventHubsConsumer
    subscribe:
      topic: !ref commandsEventHub
      fromBeginning: false

commandsMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          endpoint: !ref commandsEndpoint
        publish:
          topic: !sub '${topicRoot}/azure/event-hubs/commands'

With fromBeginning: false, the consumer starts at the end of the event hub and only receives new events. Every consumed event arrives on the MQTT topic in the following format, with the event body as a string in the message property:

{
  "timestamp": 1712345678901,
  "value": {
    "topic": "factory-commands",
    "partition": 0,
    "key": "press-01",
    "headers": {},
    "message": "{\"command\":\"stop\"}"
  }
}

Verifying the Integration

1. Install the service and set the parameters with the values from your Azure subscription.

2. Check that both connections are in the Connected state on the service details page in the Admin UI. If the connection string is wrong, the connections do not reach the connected state.

3. Publish a test message to enterprise/hamburg/assembly/line-1/press-01/telemetry, for example with an MQTT client or the Admin UI.

4. Use the Data Explorer to inspect the /res topic of the telemetry endpoint. On success, the result contains errorCode: 0. If Event Hubs rejects the write, the message contains an error property instead.

5. Open the telemetry event hub in the Azure portal and check that the event has arrived, for example with the Data Explorer feature of Event Hubs or the incoming messages metric.

6. Send a test event to the commands event hub, for example with the Data Explorer feature of Event Hubs, and check that it appears on the enterprise/azure/event-hubs/commands topic in Connectware.

Service Commissioning File Example

azure-event-hubs-example.yml

4KB

Open

azure-event-hubs-example.yml

---
# ----------------------------------------------------------------------------
# Service Commissioning File
# ----------------------------------------------------------------------------
# Copyright: Cybus GmbH
# Contact: [e-mail]
# ----------------------------------------------------------------------------
# Azure Event Hubs Integration (Example)
# ----------------------------------------------------------------------------

description: >
  Service commissioning file for the bidirectional integration between
  Connectware and Azure Event Hubs through the Apache Kafka endpoint (Example)

metadata:
  name: Azure Event Hubs Integration
  provider: cybus
  homepage: https://www.cybus.io
  version: 1.0.0

parameters:
  eventHubsNamespace:
    description: Name of your Event Hubs namespace, without the domain suffix
    type: string
    default: example-namespace

  connectionString:
    description: Connection string of your Event Hubs namespace
    type: string

  telemetryEventHub:
    description: Name of the event hub that receives shop floor data
    type: string
    default: factory-telemetry

  commandsEventHub:
    description: Name of the event hub that Connectware consumes from
    type: string
    default: factory-commands

  consumerGroupId:
    description: Kafka consumer group ID used when consuming from Event Hubs
    type: string
    default: connectware

  topicRoot:
    description: Root of the MQTT topic hierarchy
    type: string
    default: enterprise

resources:
  # Producer connection to the Kafka endpoint of the Event Hubs namespace.
  # Configuring the sasl property makes the connector connect over TLS.
  eventHubsProducer:
    type: Cybus::Connection
    properties:
      protocol: Kafka
      connection:
        brokers:
          - !sub '${eventHubsNamespace}.servicebus.windows.net:9093'
        sasl:
          mechanism: plain
          username: '$ConnectionString'
          password: !ref connectionString

  # Writes messages to the telemetry event hub
  telemetryEndpoint:
    type: Cybus::Endpoint
    properties:
      protocol: Kafka
      connection: !ref eventHubsProducer
      write:
        topic: !ref telemetryEventHub

  # Streams shop floor data from the MQTT topic hierarchy into the event hub.
  # The transform rule wraps each payload in the message format that the
  # Kafka connector expects.
  telemetryMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            topic: !sub '${topicRoot}/+site/+area/+line/+cell/telemetry'
          publish:
            endpoint: !ref telemetryEndpoint
          rules:
            - transform:
                expression: '{ "value": { "value": $string($) } }'

  # Consumer connection to the Kafka endpoint of the Event Hubs namespace.
  # A Kafka connection is either a producer or a consumer, so the consuming
  # direction needs its own connection resource.
  eventHubsConsumer:
    type: Cybus::Connection
    properties:
      protocol: Kafka
      connection:
        clientType: consumer
        groupId: !ref consumerGroupId
        brokers:
          - !sub '${eventHubsNamespace}.servicebus.windows.net:9093'
        sasl:
          mechanism: plain
          username: '$ConnectionString'
          password: !ref connectionString

  # Consumes new events from the commands event hub
  commandsEndpoint:
    type: Cybus::Endpoint
    properties:
      protocol: Kafka
      connection: !ref eventHubsConsumer
      subscribe:
        topic: !ref commandsEventHub
        fromBeginning: false

  # Publishes every consumed event to an MQTT topic
  commandsMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref commandsEndpoint
          publish:
            topic: !sub '${topicRoot}/azure/event-hubs/commands'