> For the complete documentation index, see [llms.txt](https://docs.cybus.io/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.cybus.io/2-4-2/guides/system-connectivity/messaging-event-streaming/azure-event-hubs-integration.md). # Azure Event Hubs Integration 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](https://learn.microsoft.com/en-us/azure/event-hubs/azure-event-hubs-apache-kafka-overview) 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](https://learn.microsoft.com/en-us/azure/event-hubs/event-hubs-get-connection-string) in the Azure documentation. * Access to the [Admin UI](/2-4-2/access/admin-ui.md) with sufficient [user permissions](/2-4-2/access/user-management.md). * Basic knowledge of MQTT and the Connectware [services](/2-4-2/data-flows/services.md) concept (for example, [service commissioning files](/2-4-2/data-flows/service-commissioning-files.md), [connections](/2-4-2/data-flows/service-commissioning-files/resources/cybus-connection.md), and [endpoints](/2-4-2/data-flows/service-commissioning-files/resources/cybus-endpoint.md)). ## Connectware and Azure Event Hubs Integration Azure Event Hubs exposes an Apache Kafka-compatible endpoint on every namespace, reachable at `.servicebus.windows.net` on port 9093. Connectware communicates with this endpoint through the [Kafka connector](/2-4-2/connectors/enterprise-connectors/kafka.md), 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 (`////`). The mapping subscribes with wildcards across all levels, so any machine in the hierarchy is picked up without changing the integration. {% hint style="info" %} 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](/2-4-2/guides/system-connectivity/cloud-iot-platforms/azure-iot-hub-integration.md). {% endhint %} ### 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`. {% code lineNumbers="true" %} ```yaml 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 ``` {% endcode %} ### 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](/2-4-2/connectors/enterprise-connectors/kafka/kafkaconnection.md). {% code lineNumbers="true" %} ```yaml resources: eventHubsProducer: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !sub '${eventHubsNamespace}.servicebus.windows.net:9093' sasl: mechanism: plain username: '$ConnectionString' password: !ref connectionString ``` {% endcode %} 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](/2-4-2/connectors/enterprise-connectors/kafka.md#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. {% code lineNumbers="true" %} ```yaml 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($) } }' ``` {% endcode %} 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: {% code lineNumbers="true" %} ```json { "temperature": 42.1, "pressure": 1.8, "serial": "SN-0001" } ``` {% endcode %} 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. {% hint style="info" %} 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. {% endhint %} ### 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](https://learn.microsoft.com/en-us/azure/event-hubs/apache-kafka-frequently-asked-questions) in the Azure documentation. {% code lineNumbers="true" %} ```yaml 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' ``` {% endcode %} 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: {% code lineNumbers="true" %} ```json { "timestamp": 1712345678901, "value": { "topic": "factory-commands", "partition": 0, "key": "press-01", "headers": {}, "message": "{\"command\":\"stop\"}" } } ``` {% endcode %} ## 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](/2-4-2/monitoring/data-explorer.md) 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 {% file src="/files/K8F0yColtUeFM660PEWi" %} {% code title="azure-event-hubs-example.yml" lineNumbers="true" expandable="true" %} ```yaml --- # ---------------------------------------------------------------------------- # Service Commissioning File # ---------------------------------------------------------------------------- # Copyright: Cybus GmbH # Contact: support@cybus.io # ---------------------------------------------------------------------------- # 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' ``` {% endcode %} --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.cybus.io/2-4-2/guides/system-connectivity/messaging-event-streaming/azure-event-hubs-integration.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.