NSQ and Apache Hudi Integration

Input and output integration overview

The NSQ Telegraf plugin reads metrics from the NSQD messaging system, allowing for real-time data processing and monitoring.

Writes metrics to Parquet files via Telegraf’s Parquet output plugin, preparing them for ingestion into Apache Hudi’s lakehouse architecture.

Integration details

NSQ

The NSQ plugin interfaces with NSQ, a real-time messaging platform, enabling the reading of messages from NSQD. This plugin is categorized as a service plugin, meaning it actively listens for metrics and events rather than polling them at regular intervals. With an emphasis on reliability, it prevents data loss by tracking undelivered messages until they are acknowledged by outputs. The plugin allows for configurations such as specifying NSQLookupd endpoints, topics, and channels, and it supports multiple data formats for flexibility in data handling.

Apache Hudi

This configuration leverages Telegraf’s Parquet plugin to serialize metrics into columnar Parquet files suitable for downstream ingestion by Apache Hudi. The plugin writes metrics grouped by metric name into files in a specified directory, buffering writes for efficiency and optionally rotating files on timers. It considers schema compatibility—metrics with incompatible schemas are dropped—ensuring consistency. Apache Hudi can then consume these Parquet files via tools like DeltaStreamer or Spark jobs, enabling transactional ingestion, time-travel queries, and upserts on your time series data.

Configuration

NSQ

# Read metrics from NSQD topic(s)
[[inputs.nsq_consumer]]
  ## Server option still works but is deprecated, we just prepend it to the nsqd array.
  # server = "localhost:4150"

  ## An array representing the NSQD TCP HTTP Endpoints
  nsqd = ["localhost:4150"]

  ## An array representing the NSQLookupd HTTP Endpoints
  nsqlookupd = ["localhost:4161"]
  topic = "telegraf"
  channel = "consumer"
  max_in_flight = 100

  ## 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

  ## 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"

Apache Hudi

[[outputs.parquet]]
  ## Directory to write parquet files in. If a file already exists the output
  ## will attempt to continue using the existing file.
  directory = "/var/lib/telegraf/hudi_metrics"

  ## File rotation interval (default is no rotation)
  # rotation_interval = "1h"

  ## Buffer size before writing (default is 1000 metrics)
  # buffer_size = 1000

  ## Optional: compression codec (snappy, gzip, etc.)
  # compression_codec = "snappy"

  ## When grouping metrics, each metric name goes to its own file
  ## If a metric’s schema doesn’t match the existing schema, it will be dropped

Input and output integration examples

NSQ

1. Real-Time Analytics Dashboard: Integrate this plugin with a visualization tool to create a dashboard that displays real-time metrics from various topics in NSQ. By subscribing to specific topics, users can monitor system health and application performance dynamically, allowing for immediate insights and timely responses to any anomalies.

2. Event-Driven Automation: Combine NSQ with a serverless architecture to trigger automated workflows based on incoming messages. This use case could involve processing data for machine learning models or responding to user actions in applications, thus streamlining operations and enhancing user experience through rapid processing.

3. Multi-Service Communication Hub: Use the NSQ plugin to act as a centralized messaging hub among different microservices in a distributed architecture. By enabling services to communicate through NSQ, developers can ensure reliable message delivery while maintaining decoupled service interactions, significantly improving scalability and resilience.

4. Metrics Aggregation for Enhanced Monitoring: Implement the NSQ plugin to aggregate metrics from multiple sources before sending them to an analytics tool. This setup enables businesses to consolidate data from various applications and services, creating a unified view for better decision-making and strategic planning.

Apache Hudi

1. Transactional Lakehouse Metrics: Buffer and write Web service metrics as Parquet files for DeltaStreamer to ingest into Hudi, enabling upserts, ACID compliance, and time-travel on historical performance data.

2. Edge Device Batch Analytics: Telegraf running on IoT gateways writes metrics to Parquet locally, where periodic Spark jobs ingest them into Hudi for long-term analytics and traceability.

3. Schema-Enforced Abnormal Metric Handling: Use Parquet plugin’s strict schema-dropping behavior to prevent malformed or unexpected metric changes. Hudi ingestion then guarantees consistent schema and data quality in downstream datasets.

4. Data Platform Integration: Store Telegraf metrics as Parquet files in an S3/ADLS landing zone. Hudi’s Spark-based ingestion pipeline then loads them into a unified, queryable lakehouse with business events and logs.

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.