Azure Event Hubs and Elasticsearch Integration
Powerful performance with an easy integration, powered by Telegraf, the open source data connector built by InfluxData.
5B+
Telegraf downloads
#1
Time series database
Source: DB Engines
1B+
Downloads of InfluxDB
2,800+
Contributors
Table of Contents
Powerful Performance, Limitless Scale
Collect, organize, and act on massive volumes of high-velocity data. Any data is more valuable when you think of it as time series data. with InfluxDB, the #1 time series platform built to scale with Telegraf.
See Ways to Get Started
Input and output integration overview
The Azure Event Hubs Input Plugin allows Telegraf to consume data from Azure Event Hubs and Azure IoT Hub, enabling efficient data processing and monitoring of event streams from these cloud services.
The Telegraf Elasticsearch Plugin seamlessly sends metrics to an Elasticsearch server. The plugin handles template creation and dynamic index management, and supports various Elasticsearch-specific features to ensure data is formatted correctly for storage and retrieval.
Integration details
Azure Event Hubs
This plugin serves as a consumer for Azure Event Hubs and Azure IoT Hub, allowing users to ingest data streams from these platforms efficiently. Azure Event Hubs is a highly scalable data streaming platform and event ingestion service capable of receiving and processing millions of events per second, while Azure IoT Hub enables secure device-to-cloud and cloud-to-device communication in IoT applications. The Event Hub Input Plugin interacts seamlessly with these services, providing reliable message consumption and stream processing capabilities. Key features include dynamic management of consumer groups, message tracking to prevent data loss, and customizable settings for prefetch counts, user agents, and metadata handling. This plugin is designed to support a range of use cases, including real-time telemetry data collection, IoT data processing, and integration with various data analysis and monitoring tools within the broader Azure ecosystem.
Elasticsearch
This plugin writes metrics to Elasticsearch, a distributed, RESTful search and analytics engine capable of storing large amounts of data in near real-time. It is designed to handle Elasticsearch versions 5.x through 7.x and utilizes its dynamic template features to manage data type mapping properly. The plugin supports advanced features such as template management, dynamic index naming, and integration with OpenSearch. It also allows configurations for authentication and health monitoring of the Elasticsearch nodes.
Configuration
Azure Event Hubs
[[inputs.eventhub_consumer]]
## The default behavior is to create a new Event Hub client from environment variables.
## This requires one of the following sets of environment variables to be set:
##
## 1) Expected Environment Variables:
## - "EVENTHUB_CONNECTION_STRING"
##
## 2) Expected Environment Variables:
## - "EVENTHUB_NAMESPACE"
## - "EVENTHUB_NAME"
## - "EVENTHUB_KEY_NAME"
## - "EVENTHUB_KEY_VALUE"
## 3) Expected Environment Variables:
## - "EVENTHUB_NAMESPACE"
## - "EVENTHUB_NAME"
## - "AZURE_TENANT_ID"
## - "AZURE_CLIENT_ID"
## - "AZURE_CLIENT_SECRET"
## Uncommenting the option below will create an Event Hub client based solely on the connection string.
## This can either be the associated environment variable or hard coded directly.
## If this option is uncommented, environment variables will be ignored.
## Connection string should contain EventHubName (EntityPath)
# connection_string = ""
## Set persistence directory to a valid folder to use a file persister instead of an in-memory persister
# persistence_dir = ""
## Change the default consumer group
# consumer_group = ""
## By default the event hub receives all messages present on the broker, alternative modes can be set below.
## The timestamp should be in https://github.com/toml-lang/toml#offset-date-time format (RFC 3339).
## The 3 options below only apply if no valid persister is read from memory or file (e.g. first run).
# from_timestamp =
# latest = true
## Set a custom prefetch count for the receiver(s)
# prefetch_count = 1000
## Add an epoch to the receiver(s)
# epoch = 0
## Change to set a custom user agent, "telegraf" is used by default
# user_agent = "telegraf"
## To consume from a specific partition, set the partition_ids option.
## An empty array will result in receiving from all partitions.
# partition_ids = ["0","1"]
## Max undelivered messages
## This plugin uses tracking metrics, which ensure messages are read to
## outputs before acknowledging them to the original broker to ensure data
## is not lost. This option sets the maximum messages to read from the
## broker that have not been written by an output.
##
## This value needs to be picked with awareness of the agent's
## metric_batch_size value as well. Setting max undelivered messages too high
## can result in a constant stream of data batches to the output. While
## setting it too low may never flush the broker's messages.
# max_undelivered_messages = 1000
## Set either option below to true to use a system property as timestamp.
## You have the choice between EnqueuedTime and IoTHubEnqueuedTime.
## It is recommended to use this setting when the data itself has no timestamp.
# enqueued_time_as_ts = true
# iot_hub_enqueued_time_as_ts = true
## Tags or fields to create from keys present in the application property bag.
## These could for example be set by message enrichments in Azure IoT Hub.
# application_property_tags = []
# application_property_fields = []
## Tag or field name to use for metadata
## By default all metadata is disabled
# sequence_number_field = "SequenceNumber"
# enqueued_time_field = "EnqueuedTime"
# offset_field = "Offset"
# partition_id_tag = "PartitionID"
# partition_key_tag = "PartitionKey"
# iot_hub_device_connection_id_tag = "IoTHubDeviceConnectionID"
# iot_hub_auth_generation_id_tag = "IoTHubAuthGenerationID"
# iot_hub_connection_auth_method_tag = "IoTHubConnectionAuthMethod"
# iot_hub_connection_module_id_tag = "IoTHubConnectionModuleID"
# iot_hub_enqueued_time_field = "IoTHubEnqueuedTime"
## Data format to consume.
## Each data format has its own unique set of configuration options, read
## more about them here:
## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
data_format = "influx"
Elasticsearch
[[outputs.elasticsearch]]
## The full HTTP endpoint URL for your Elasticsearch instance
## Multiple urls can be specified as part of the same cluster,
## this means that only ONE of the urls will be written to each interval
urls = [ "http://node1.es.example.com:9200" ] # required.
## Elasticsearch client timeout, defaults to "5s" if not set.
timeout = "5s"
## Set to true to ask Elasticsearch a list of all cluster nodes,
## thus it is not necessary to list all nodes in the urls config option
enable_sniffer = false
## Set to true to enable gzip compression
enable_gzip = false
## Set the interval to check if the Elasticsearch nodes are available
## Setting to "0s" will disable the health check (not recommended in production)
health_check_interval = "10s"
## Set the timeout for periodic health checks.
# health_check_timeout = "1s"
## HTTP basic authentication details.
## HTTP basic authentication details
# username = "telegraf"
# password = "mypassword"
## HTTP bearer token authentication details
# auth_bearer_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"
## Index Config
## The target index for metrics (Elasticsearch will create if it not exists).
## You can use the date specifiers below to create indexes per time frame.
## The metric timestamp will be used to decide the destination index name
# %Y - year (2016)
# %y - last two digits of year (00..99)
# %m - month (01..12)
# %d - day of month (e.g., 01)
# %H - hour (00..23)
# %V - week of the year (ISO week) (01..53)
## Additionally, you can specify a tag name using the notation {{tag_name}}
## which will be used as part of the index name. If the tag does not exist,
## the default tag value will be used.
# index_name = "telegraf-{{host}}-%Y.%m.%d"
# default_tag_value = "none"
index_name = "telegraf-%Y.%m.%d" # required.
## Optional Index Config
## Set to true if Telegraf should use the "create" OpType while indexing
# use_optype_create = false
## Optional TLS Config
# tls_ca = "/etc/telegraf/ca.pem"
# tls_cert = "/etc/telegraf/cert.pem"
# tls_key = "/etc/telegraf/key.pem"
## Use TLS but skip chain & host verification
# insecure_skip_verify = false
## Template Config
## Set to true if you want telegraf to manage its index template.
## If enabled it will create a recommended index template for telegraf indexes
manage_template = true
## The template name used for telegraf indexes
template_name = "telegraf"
## Set to true if you want telegraf to overwrite an existing template
overwrite_template = false
## If set to true a unique ID hash will be sent as sha256(concat(timestamp,measurement,series-hash)) string
## it will enable data resend and update metric points avoiding duplicated metrics with different id's
force_document_id = false
## Specifies the handling of NaN and Inf values.
## This option can have the following values:
## none -- do not modify field-values (default); will produce an error if NaNs or infs are encountered
## drop -- drop fields containing NaNs or infs
## replace -- replace with the value in "float_replacement_value" (default: 0.0)
## NaNs and inf will be replaced with the given number, -inf with the negative of that number
# float_handling = "none"
# float_replacement_value = 0.0
## Pipeline Config
## To use a ingest pipeline, set this to the name of the pipeline you want to use.
# use_pipeline = "my_pipeline"
## Additionally, you can specify a tag name using the notation {{tag_name}}
## which will be used as part of the pipeline name. If the tag does not exist,
## the default pipeline will be used as the pipeline. If no default pipeline is set,
## no pipeline is used for the metric.
# use_pipeline = "{{es_pipeline}}"
# default_pipeline = "my_pipeline"
#
# Custom HTTP headers
# To pass custom HTTP headers please define it in a given below section
# [outputs.elasticsearch.headers]
# "X-Custom-Header" = "custom-value"
## Template Index Settings
## Overrides the template settings.index section with any provided options.
## Defaults provided here in the config
# template_index_settings = {
# refresh_interval = "10s",
# mapping.total_fields.limit = 5000,
# auto_expand_replicas = "0-1",
# codec = "best_compression"
# }
Input and output integration examples
Azure Event Hubs
-
Real-Time IoT Device Monitoring: Use the Azure Event Hubs Plugin to monitor telemetry data from IoT devices like sensors and actuators. By streaming device data into monitoring dashboards, organizations can gain insights into system performances, track usage patterns, and quickly respond to irregularities. This setup allows for proactive management of devices, improving operational efficiency and reducing downtime.
-
Event-Driven Data Processing Workflows: Leverage this plugin to trigger data processing workflows in response to events received from Azure Event Hubs. For instance, when a new event arrives, it can initiate data transformation, aggregation, or storage processes, allowing businesses to automate their workflows more effectively. This integration enhances responsiveness and streamlines operations across systems.
-
Integration with Analytics Platforms: Implement the plugin to funnel event data into analytics platforms like Azure Synapse or Power BI. By integrating real-time streaming data into analytics tools, organizations can perform comprehensive data analysis, drive business intelligence efforts, and create interactive visualizations that inform decision-making.
-
Cross-Platform Data Sync: Utilize the Azure Event Hubs Plugin to synchronize data streams across diverse systems or platforms. By consuming data from Azure Event Hubs and forwarding it to other systems like databases or cloud storage, organizations can maintain consistent and up-to-date information across their entire architecture, enabling cohesive data strategies.
Elasticsearch
-
Time-based Indexing: Use this plugin to store metrics in Elasticsearch to index each metric based on the time collected. For example, CPU metrics can be stored in a daily index named
telegraf-2023.01.01
, allowing easy time-based queries and retention policies. -
Dynamic Templates Management: Utilize the template management feature to automatically create a custom template tailored to your metrics. This allows you to define how different fields are indexed and analyzed without manually configuring Elasticsearch, ensuring an optimal data structure for querying.
-
OpenSearch Compatibility: If you are using AWS OpenSearch, you can configure this plugin to work seamlessly by activating compatibility mode, ensuring your existing Elasticsearch clients remain functional and compatible with newer cluster setups.
Feedback
Thank you for being part of our community! If you have any general feedback or found any bugs on these pages, we welcome and encourage your input. Please submit your feedback in the InfluxDB community Slack.
Powerful Performance, Limitless Scale
Collect, organize, and act on massive volumes of high-velocity data. Any data is more valuable when you think of it as time series data. with InfluxDB, the #1 time series platform built to scale with Telegraf.
See Ways to Get Started
Related Integrations
Related Integrations
HTTP and InfluxDB Integration
The HTTP plugin collects metrics from one or more HTTP(S) endpoints. It supports various authentication methods and configuration options for data formats.
View IntegrationKafka and InfluxDB Integration
This plugin reads messages from Kafka and allows the creation of metrics based on those messages. It supports various configurations including different Kafka settings and message processing options.
View IntegrationKinesis and InfluxDB Integration
The Kinesis plugin allows for reading metrics from AWS Kinesis streams. It supports multiple input data formats and offers checkpointing features with DynamoDB for reliable message processing.
View Integration