Kafka 和 MongoDB 集成

强大的性能和简易的集成,由 InfluxData 构建的开源数据连接器 Telegraf 提供支持。

info

这不是大规模实时查询的推荐配置。为了优化查询和压缩、高速摄取和高可用性,您可能需要考虑 Kafka 和 InfluxDB 集成

5B+

Telegraf 下载量

#1

时序数据库
来源:DB Engines

1B+

InfluxDB 下载量

2,800+

贡献者

目录

强大的性能,无限的扩展能力

收集、组织和处理海量高速数据。当您将任何数据视为时序数据时,它都会变得更有价值。InfluxDB 是排名第一的时序平台,旨在与 Telegraf 一起扩展。

查看入门方法

输入和输出集成概述

此插件允许您实时从 Kafka 主题收集指标,从而增强 Telegraf 设置中的数据监控和收集能力。

MongoDB Telegraf 插件使用户能够将指标发送到 MongoDB 数据库,并自动管理时序集合。

集成详情

Kafka

Kafka Telegraf 插件旨在从 Kafka 主题读取数据,并使用支持的输入数据格式创建指标。作为服务输入插件,它持续监听传入的指标和事件,这与以固定间隔运行的标准输入插件不同。此特定插件可以使用各种 Kafka 版本的功能,并能够从指定主题消费消息,应用诸如使用 SASL 的安全凭证等配置,以及使用消息偏移量和消费者组等选项管理消息处理。此插件的灵活性使其能够处理各种消息格式和用例,使其成为依赖 Kafka 进行数据摄取的应用程序的宝贵资产。

MongoDB

此插件将指标发送到 MongoDB,并与其时序功能无缝集成,从而允许在时序集合尚不存在时自动创建为时序集合。它需要 MongoDB 5.0 或更高版本才能使用时序集合功能,这对于高效存储和查询基于时间的数据至关重要。此插件通过确保所有相关指标都正确存储和组织在 MongoDB 中,从而增强了监控功能,使用户能够利用 MongoDB 强大的查询和聚合功能进行时序分析。

配置

Kafka


[[inputs.kafka_consumer]]
              ## Kafka brokers.
              brokers = ["localhost:9092"]

              ## Set the minimal supported Kafka version. Should be a string contains
              ## 4 digits in case if it is 0 version and 3 digits for versions starting
              ## from 1.0.0 separated by dot. This setting enables the use of new
              ## Kafka features and APIs.  Must be 0.10.2.0(used as default) or greater.
              ## Please, check the list of supported versions at
              ## https://pkg.go.dev/github.com/Shopify/sarama#SupportedVersions
              ##   ex: kafka_version = "2.6.0"
              ##   ex: kafka_version = "0.10.2.0"
              # kafka_version = "0.10.2.0"

              ## Topics to consume.
              topics = ["telegraf"]

              ## Topic regular expressions to consume.  Matches will be added to topics.
              ## Example: topic_regexps = [ "*test", "metric[0-9A-z]*" ]
              # topic_regexps = [ ]

              ## When set this tag will be added to all metrics with the topic as the value.
              # topic_tag = ""

              ## The list of Kafka message headers that should be pass as metric tags
              ## works only for Kafka version 0.11+, on lower versions the message headers
              ## are not available
              # msg_headers_as_tags = []

              ## The name of kafka message header which value should override the metric name.
              ## In case when the same header specified in current option and in msg_headers_as_tags
              ## option, it will be excluded from the msg_headers_as_tags list.
              # msg_header_as_metric_name = ""

              ## Set metric(s) timestamp using the given source.
              ## Available options are:
              ##   metric -- do not modify the metric timestamp
              ##   inner  -- use the inner message timestamp (Kafka v0.10+)
              ##   outer  -- use the outer (compressed) block timestamp (Kafka v0.10+)
              # timestamp_source = "metric"

              ## Optional Client id
              # client_id = "Telegraf"

              ## Optional TLS Config
              # enable_tls = false
              # 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

              ## Period between keep alive probes.
              ## Defaults to the OS configuration if not specified or zero.
              # keep_alive_period = "15s"

              ## SASL authentication credentials.  These settings should typically be used
              ## with TLS encryption enabled
              # sasl_username = "kafka"
              # sasl_password = "secret"

              ## Optional SASL:
              ## one of: OAUTHBEARER, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI
              ## (defaults to PLAIN)
              # sasl_mechanism = ""

              ## used if sasl_mechanism is GSSAPI
              # sasl_gssapi_service_name = ""
              # ## One of: KRB5_USER_AUTH and KRB5_KEYTAB_AUTH
              # sasl_gssapi_auth_type = "KRB5_USER_AUTH"
              # sasl_gssapi_kerberos_config_path = "/"
              # sasl_gssapi_realm = "realm"
              # sasl_gssapi_key_tab_path = ""
              # sasl_gssapi_disable_pafxfast = false

              ## used if sasl_mechanism is OAUTHBEARER
              # sasl_access_token = ""

              ## SASL protocol version.  When connecting to Azure EventHub set to 0.
              # sasl_version = 1

              # Disable Kafka metadata full fetch
              # metadata_full = false

              ## Name of the consumer group.
              # consumer_group = "telegraf_metrics_consumers"

              ## Compression codec represents the various compression codecs recognized by
              ## Kafka in messages.
              ##  0 : None
              ##  1 : Gzip
              ##  2 : Snappy
              ##  3 : LZ4
              ##  4 : ZSTD
              # compression_codec = 0
              ## Initial offset position; one of "oldest" or "newest".
              # offset = "oldest"

              ## Consumer group partition assignment strategy; one of "range", "roundrobin" or "sticky".
              # balance_strategy = "range"

              ## Maximum number of retries for metadata operations including
              ## connecting. Sets Sarama library's Metadata.Retry.Max config value. If 0 or
              ## unset, use the Sarama default of 3,
              # metadata_retry_max = 0

              ## Type of retry backoff. Valid options: "constant", "exponential"
              # metadata_retry_type = "constant"

              ## Amount of time to wait before retrying. When metadata_retry_type is
              ## "constant", each retry is delayed this amount. When "exponential", the
              ## first retry is delayed this amount, and subsequent delays are doubled. If 0
              ## or unset, use the Sarama default of 250 ms
              # metadata_retry_backoff = 0

              ## Maximum amount of time to wait before retrying when metadata_retry_type is
              ## "exponential". Ignored for other retry types. If 0, there is no backoff
              ## limit.
              # metadata_retry_max_duration = 0

              ## When set to true, this turns each bootstrap broker address into a set of
              ## IPs, then does a reverse lookup on each one to get its canonical hostname.
              ## This list of hostnames then replaces the original address list.
              ## resolve_canonical_bootstrap_servers_only = false

              ## Strategy for making connection to kafka brokers. Valid options: "startup",
              ## "defer". If set to "defer" the plugin is allowed to start before making a
              ## connection. This is useful if the broker may be down when telegraf is
              ## started, but if there are any typos in the broker setting, they will cause
              ## connection failures without warning at startup
              # connection_strategy = "startup"

              ## Maximum length of a message to consume, in bytes (default 0/unlimited);
              ## larger messages are dropped
              max_message_len = 1000000

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

              ## Maximum amount of time the consumer should take to process messages. If
              ## the debug log prints messages from sarama about 'abandoning subscription
              ## to [topic] because consuming was taking too long', increase this value to
              ## longer than the time taken by the output plugin(s).
              ##
              ## Note that the effective timeout could be between 'max_processing_time' and
              ## '2 * max_processing_time'.
              # max_processing_time = "100ms"

              ## The default number of message bytes to fetch from the broker in each
              ## request (default 1MB). This should be larger than the majority of
              ## your messages, or else the consumer will spend a lot of time
              ## negotiating sizes and not actually consuming. Similar to the JVM's
              ## `fetch.message.max.bytes`.
              # consumer_fetch_default = "1MB"

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

MongoDB

[[outputs.mongodb]]
              # connection string examples for mongodb
              dsn = "mongodb://localhost:27017"
              # dsn = "mongodb://mongod1:27017,mongod2:27017,mongod3:27017/admin&replicaSet=myReplSet&w=1"

              # overrides serverSelectionTimeoutMS in dsn if set
              # timeout = "30s"

              # default authentication, optional
              # authentication = "NONE"

              # for SCRAM-SHA-256 authentication
              # authentication = "SCRAM"
              # username = "root"
              # password = "***"

              # for x509 certificate authentication
              # authentication = "X509"
              # tls_ca = "ca.pem"
              # tls_key = "client.pem"
              # # tls_key_pwd = "changeme" # required for encrypted tls_key
              # insecure_skip_verify = false

              # database to store measurements and time series collections
              # database = "telegraf"

              # granularity can be seconds, minutes, or hours.
              # configuring this value will be based on your input collection frequency.
              # see https://docs.mongodb.com/manual/core/timeseries-collections/#create-a-time-series-collection
              # granularity = "seconds"

              # optionally set a TTL to automatically expire documents from the measurement collections.
              # ttl = "360h"

输入和输出集成示例

Kafka

  1. 实时数据处理:使用 Kafka 插件将来自 Kafka 主题的实时数据馈送到监控系统。这对于需要即时反馈性能指标或用户活动的应用尤其有用,使企业能够更快速地对其环境中的变化条件做出反应。

  2. 动态指标收集:利用此插件根据 Kafka 中发生的事件动态调整正在捕获的指标。例如,通过与其他服务集成,用户可以让插件即时重新配置自身,确保始终根据业务或应用程序的需求收集相关指标。

  3. 集中式日志记录和监控:使用 Kafka Consumer 插件实施集中式日志记录系统,以将来自多个服务的日志聚合到统一的监控仪表板中。此设置可以帮助识别不同服务之间的问题,并提高整体系统可观察性和故障排除能力。

  4. 异常检测系统:将 Kafka 与机器学习算法结合使用,以进行实时异常检测。通过不断分析流数据,此设置可以自动识别异常模式,触发警报并更有效地缓解潜在问题。

MongoDB

  1. 物联网设备的动态日志记录到 MongoDB:利用此插件实时收集和存储来自大量物联网设备的指标。通过将设备日志直接发送到 MongoDB,您可以创建一个集中式数据库,从而可以轻松访问和查询健康指标和性能数据,从而可以根据历史趋势进行主动维护和故障排除。

  2. Web 流量的时序分析:使用 MongoDB Telegraf 插件收集和分析一段时间内的 Web 流量指标。此应用可以帮助您了解高峰使用时间、用户互动和行为模式,从而指导营销策略和基础设施扩展决策,以改善用户体验。

  3. 自动化监控和警报系统:将 MongoDB 插件集成到跟踪应用程序性能指标的自动化监控系统中。借助时序集合,您可以根据特定阈值设置警报,从而使您的团队能够在潜在问题影响用户之前做出响应。这种主动管理可以提高服务可靠性和整体性能。

  4. 指标存储中的数据保留和 TTL 管理:利用 MongoDB 集合中文档的 TTL 功能来自动过期过时的指标。这对于仅最近的性能数据相关的环境尤其有用,可以防止您的 MongoDB 数据库因旧指标而变得混乱,并确保高效的数据管理。

反馈

感谢您成为我们社区的一份子!如果您有任何一般性反馈或在这些页面上发现了任何错误,我们欢迎并鼓励您提出意见。请在 InfluxDB 社区 Slack 中提交您的反馈。

强大的性能,无限的扩展能力

收集、组织和处理海量高速数据。当您将任何数据视为时序数据时,它都会变得更有价值。InfluxDB 是排名第一的时序平台,旨在与 Telegraf 一起扩展。

查看入门方法

相关集成

HTTP 和 InfluxDB 集成

HTTP 插件从一个或多个 HTTP(S) 端点收集指标。它支持各种身份验证方法和数据格式的配置选项。

查看集成

Kafka 和 InfluxDB 集成

此插件从 Kafka 读取消息,并允许基于这些消息创建指标。它支持各种配置,包括不同的 Kafka 设置和消息处理选项。

查看集成

Kinesis 和 InfluxDB 集成

Kinesis 插件允许从 AWS Kinesis 流中读取指标。它支持多种输入数据格式,并提供与 DynamoDB 的检查点功能,以实现可靠的消息处理。

查看集成