ConsumeAzureEventHub

Description:

Receives messages from Microsoft Azure Event Hubs with checkpointing to ensure consistent event processing. Checkpoint tracking avoids consuming a message multiple times and enables reliable resumption of processing in the event of intermittent network failures. Checkpoint tracking requires external storage and provides the preferred approach to consuming messages from Azure Event Hubs. In clustered environment, ConsumeAzureEventHub processor instances form a consumer group and the messages are distributed among the cluster nodes (each message is processed on one cluster node only).

Tags:

azure, microsoft, cloud, eventhub, events, streaming, streams

Properties:

In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.

Display NameAPI NameDefault ValueAllowable ValuesDescription
Event Hub Namespaceevent-hub-namespaceThe namespace that the Azure Event Hubs is assigned to. This is generally equal to <Event Hub Names>-ns.
Supports Expression Language: true (will be evaluated using variable registry only)
Event Hub Nameevent-hub-nameThe name of the event hub to pull messages from.
Supports Expression Language: true (will be evaluated using variable registry only)
Service Bus EndpointService Bus EndpointAzure
  • Azure Servicebus endpoint for general use
  • Azure China Servicebus endpoint for China
  • Azure Germany Servicebus endpoint for Germany
  • Azure US Government Servicebus endpoint for US Government
To support namespaces not in the default windows.net domain.
Transport TypeTransport TypeAMQP
  • AMQP AMQP over TCP on ports 5671 and 5672
  • AMQP_WEB_SOCKETS AMQP over HTTPS with WebSockets on port 443
Advanced Message Queuing Protocol Transport Type for communication with Azure Event Hubs
Shared Access Policy Nameevent-hub-shared-access-policy-nameThe name of the shared access policy. This policy must have Listen claims.
Supports Expression Language: true (will be evaluated using variable registry only)
Shared Access Policy Keyevent-hub-shared-access-policy-primary-keyThe key of the shared access policy. Either the primary or the secondary key can be used.
Sensitive Property: true
Use Azure Managed Identityuse-managed-identityfalse
  • true
  • false
Choose whether or not to use the managed identity of Azure VM/VMSS
Consumer Groupevent-hub-consumer-group$DefaultThe name of the consumer group to use.
Supports Expression Language: true (will be evaluated using variable registry only)
Consumer Hostnameevent-hub-consumer-hostnameDEPRECATED: This property is no longer used.
Supports Expression Language: true (will be evaluated using variable registry only)
Record Readerrecord-readerController Service API:
RecordReaderFactory
Implementations: GrokReader
JsonTreeReader
WindowsEventLogReader
ReaderLookup
ParquetReader
CSVReader
Syslog5424Reader
ExcelReader
CEFReader
XMLReader
ScriptedReader
SyslogReader
JsonPathReader
AvroReader
YamlTreeReader
The Record Reader to use for reading received messages. The event hub name can be referred by Expression Language '${eventhub.name}' to access a schema.
Record Writerrecord-writerController Service API:
RecordSetWriterFactory
Implementations: FreeFormTextRecordSetWriter
CSVRecordSetWriter
ParquetRecordSetWriter
RecordSetWriterLookup
ScriptedRecordSetWriter
XMLRecordSetWriter
JsonRecordSetWriter
AvroRecordSetWriter
The Record Writer to use for serializing Records to an output FlowFile. The event hub name can be referred by Expression Language '${eventhub.name}' to access a schema. If not specified, each message will create a FlowFile.
Initial Offsetevent-hub-initial-offsetEnd of stream
  • Start of stream Read from the oldest message retained in the stream.
  • End of stream Ignore old retained messages even if exist, start reading new ones from now.
Specify where to start receiving messages if offset is not yet stored in the checkpoint store.
Prefetch Countevent-hub-prefetch-count300No Description Provided.
Supports Expression Language: true (will be evaluated using variable registry only)
Batch Sizeevent-hub-batch-size10The number of messages to process within a NiFi session. This parameter affects throughput and consistency. NiFi commits its session and Event Hubs checkpoints after processing this number of messages. If NiFi session is committed, but fails to create an Event Hubs checkpoint, then it is possible that the same messages will be received again. The higher number, the higher throughput, but possibly less consistent.
Supports Expression Language: true (will be evaluated using variable registry only)
Message Receive Timeoutevent-hub-message-receive-timeout1 minThe amount of time this consumer should wait to receive the Batch Size before returning.
Supports Expression Language: true (will be evaluated using variable registry only)
Checkpoint Strategycheckpoint-strategyAzure Blob Storage
  • Azure Blob Storage Use Azure Blob Storage to store partition ownership and checkpoint information
  • Component State Use component state to store partition ownership and checkpoint information
Specifies which strategy to use for storing and retrieving partition ownership and checkpoint information for each partition.
Storage Account Namestorage-account-nameName of the Azure Storage account to store event hub consumer group state.
Supports Expression Language: true (will be evaluated using variable registry only)

This Property is only considered if the [Checkpoint Strategy] Property has a value of "Azure Blob Storage".
Storage Account Keystorage-account-keyThe Azure Storage account key to store event hub consumer group state.
Sensitive Property: true
Supports Expression Language: true (will be evaluated using variable registry only)

This Property is only considered if the [Checkpoint Strategy] Property has a value of "Azure Blob Storage".
Storage SAS Tokenstorage-sas-tokenThe Azure Storage SAS token to store Event Hub consumer group state. Always starts with a ? character.
Sensitive Property: true
Supports Expression Language: true (will be evaluated using variable registry only)

This Property is only considered if the [Checkpoint Strategy] Property has a value of "Azure Blob Storage".
Storage Container Namestorage-container-nameName of the Azure Storage container to store the event hub consumer group state. If not specified, event hub name is used.
Supports Expression Language: true (will be evaluated using variable registry only)

This Property is only considered if the [Checkpoint Strategy] Property has a value of "Azure Blob Storage".
Proxy Configuration Serviceproxy-configuration-serviceController Service API:
ProxyConfigurationService
Implementation: StandardProxyConfigurationService
Specifies the Proxy Configuration Controller Service to proxy network requests. Supported proxies: HTTP + AuthN

This Property is only considered if the [Transport Type] Property has a value of "AMQP_WEB_SOCKETS".

Relationships:

NameDescription
successFlowFiles received from Event Hub.
parse.failureIf a message from event hub cannot be parsed using the configured Record Reader or failed to be written by the configured Record Writer, the contents of the message will be routed to this Relationship as its own individual FlowFile.

Reads Attributes:

None specified.

Writes Attributes:

NameDescription
eventhub.enqueued.timestampThe time (in milliseconds since epoch, UTC) at which the message was enqueued in the event hub
eventhub.offsetThe offset into the partition at which the message was stored
eventhub.sequenceThe sequence number associated with the message
eventhub.nameThe name of the event hub from which the message was pulled
eventhub.partitionThe name of the partition from which the message was pulled
eventhub.property.*The application properties of this message. IE: 'application' would be 'eventhub.property.application'

State management:

ScopeDescription
LOCAL, CLUSTERLocal state is used to store the client id. Cluster state is used to store partition ownership and checkpoint information when component state is configured as the checkpointing strategy.

Restricted:

This component is not restricted.

Input requirement:

This component does not allow an incoming relationship.

System Resource Considerations:

None specified.