Skip to main content

Azure IoT Hub Source Node

The azureiothubsource node consumes device-to-cloud telemetry from an Azure IoT Hub through the hub's built-in Event Hub-compatible endpoint. Beyond decoding the message body, it attaches IoT-specific metadata — the originating device id, the enqueued time, and any application properties — under a reserved _iothub key on each record.

Mechanically this node shares the Event Hub consumer machinery: partition balancing, blob checkpointing, and the same authentication rules as azureeventhubsource. What differs is where the connection details come from and the metadata envelope it adds.

Key Features​

  • Device attribution: the originating device id is extracted from IoT Hub's iothub-connection-device-id system property and surfaced as _iothub.deviceId
  • Application properties preserved: user-defined message properties are nested under _iothub.properties
  • Body fields stay top-level: the telemetry payload's own fields remain at the record root, so downstream nodes address them directly
  • Automatic partition balancing: partitions of the built-in endpoint are claimed cooperatively across workers
  • Durable checkpointing: partition offsets live in an Azure Blob Storage container, so restarts resume from the last committed offset
  • Backpressure: a bounded buffer between Azure's delivery callback and the pull loop blocks delivery rather than growing memory
  • Infinite streaming: never exhausts — runs until the pipeline is terminated

Configuration​

FieldTypeRequiredDefaultDescription
eventHubNameStringYes—The Event Hub-compatible name of the IoT Hub built-in endpoint — not the IoT Hub's own name
consumerGroupStringYes$DefaultConsumer group on the built-in endpoint
connectionStringStringConditional—Event Hub-compatible connection string for the built-in endpoint. Mutually exclusive with fullyQualifiedNamespace
fullyQualifiedNamespaceStringConditional—Event Hub-compatible namespace host for Entra ID auth. Mutually exclusive with connectionString
tenantIdStringNo—Entra ID service principal tenant. Requires clientId, clientSecret and fullyQualifiedNamespace
clientIdStringNo—Entra ID service principal client id
clientSecretStringNo—Entra ID service principal secret
checkpointContainerNameStringYes—Blob container holding checkpoint and partition-ownership data
checkpointStorageConnectionStringStringConditional—Connection string for the checkpoint storage account. Mutually exclusive with checkpointStorageEndpoint
checkpointStorageEndpointStringConditional—Blob endpoint for the checkpoint storage account, authenticated with Entra ID. Mutually exclusive with checkpointStorageConnectionString
encodingTypeStringYes—Encoding of the telemetry message body. See Encoding Types
initialPositionStringNoEARLIESTWhere to start when a partition has no persisted checkpoint: EARLIEST or LATEST. Ignored once a checkpoint exists
maxBufferedEventsintNo1024Bound on events buffered between Azure's callback and the pull loop
maxEventsPerFetchintNo500Maximum events drained from the buffer per fetch cycle
commitStrategyStringNoBATCHCheckpoint strategy: PER_RECORD, BATCH, or NONE
commitBatchSizeintNo1000Records between checkpoints when commitStrategy is BATCH
commitIntervalMslongNo5000Milliseconds between checkpoints when commitStrategy is BATCH

Finding the Event Hub-compatible endpoint​

IoT Hub does not expose itself as an Event Hub under its own name. The values this node needs live in the Azure portal under IoT Hub → Hub settings → Built-in endpoints:

  • Event Hub-compatible name → eventHubName
  • Event Hub-compatible endpoint / connection string → connectionString

Using the IoT Hub's resource name or its service connection string will not work.

Record Shape​

The telemetry body is deserialized with encodingType and its fields stay at the top level. IoT metadata is nested under _iothub:

{
  "temperature": 21.4,
  "humidity": 38,
  "_iothub": {
    "deviceId": "sensor-42",
    "enqueuedTime": "2026-07-23T14:02:11.318Z",
    "properties": {
      "messageType": "telemetry",
      "schemaVersion": "2"
    }
  }
}

Notes on the envelope:

  • _iothub is present only when the message carried at least one piece of metadata. A message with none produces a record with no envelope at all.
  • When the envelope is present, _iothub.properties is always present — an empty object if the message had no application properties. Downstream nodes can rely on that.
  • All metadata values are strings, including enqueuedTime (ISO-8601) and any numeric application properties.
  • A body field literally named _iothub is overwritten by the envelope.

How It Works​

The node runs Azure's EventProcessorClient against the built-in endpoint and bridges its push-based delivery onto ZephFlow's pull-based source contract:

  1. The processor client claims a share of the endpoint's partitions and delivers events by callback.
  2. Callbacks enqueue events into a bounded buffer (maxBufferedEvents); a full buffer blocks delivery, applying backpressure to IoT Hub.
  3. The polling thread drains up to maxEventsPerFetch events, deserializes each body, builds the _iothub envelope from the event's system and application properties, and merges it into each resulting record.
  4. On commit, the offset of the latest drained event per partition is checkpointed to the blob container.

Authentication, checkpoint store setup, and commit strategy behave exactly as documented for azureeventhubsource. Delivery is at-least-once.

A message whose body fails to deserialize is counted as a deserialization failure and routed to the DLQ when dlqConfig is set.

DAG Example​

jobContext:
  otherProperties: {}
  metricTags: {}
  dlqConfig:

dag:
  - id: "source"
    commandName: "azureiothubsource"
    config:
      connectionString: "Endpoint=sb://ihsuprodblres000000000000.servicebus.windows.net/;SharedAccessKeyName=iothubowner;SharedAccessKey=...;EntityPath=iothub-ehub-my-hub-000000-0000000000"
      eventHubName: "iothub-ehub-my-hub-000000-0000000000"
      consumerGroup: "$Default"
      checkpointStorageConnectionString: "DefaultEndpointsProtocol=https;AccountName=mystorage;AccountKey=..."
      checkpointContainerName: "iothub-checkpoints"
      encodingType: "JSON_OBJECT"
      initialPosition: "EARLIEST"
      commitStrategy: "BATCH"
    outputs:
      - "hot"

  - id: "hot"
    commandName: "filter"
    config:
      expression: $.temperature > 30
    outputs:
      - "sink"

  - id: "sink"
    commandName: "stdout"
    config:
      encodingType: "JSON_OBJECT"

Azure Setup​

  1. Enable the built-in endpoint (on by default) and read its Event Hub-compatible name and connection string from Hub settings → Built-in endpoints.
  2. Create a consumer group on the built-in endpoint for this pipeline. IoT Hub supports a limited number of consumer groups per hub, and sharing one across independent consumers causes them to fight over offsets.
  3. Create a storage account and blob container for checkpoints.
  4. Grant permissions: the identity needs service-connect rights on the IoT Hub built-in endpoint, and read/write on the checkpoint container (Storage Blob Data Contributor for Entra ID).