Kinesis 和 OpenSearch 集成

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

info

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

50 亿+

Telegraf 下载量

#1

时序数据库
来源:DB-Engines

10 亿+

InfluxDB 下载量

2,800+

贡献者

目录

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

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

查看入门方法

输入和输出集成概述

Kinesis 插件使您能够从 Kinesis 数据流中读取数据,支持各种数据格式和配置。

OpenSearch 输出插件允许用户使用 HTTP 将指标直接发送到 OpenSearch 实例,从而促进 OpenSearch 生态系统内有效的数据管理和分析。

集成详情

Kinesis

Kinesis Telegraf 插件旨在从 Amazon Kinesis 数据流中读取数据,使用户能够实时收集指标。 作为服务输入插件,它通过监听传入数据而不是定期轮询来运行。 配置指定了各种选项,包括 AWS 区域、流名称、身份验证凭据和数据格式。 它支持跟踪未送达的消息以防止数据丢失,并且用户可以利用 DynamoDB 来维护上次处理记录的检查点。 此插件对于需要可靠且可扩展的流处理以及其他监控需求的应用程序特别有用。

OpenSearch

OpenSearch Telegraf 插件通过 HTTP 与 OpenSearch 数据库集成,从而实现指标的简化收集和存储。 作为一个专为 2.x 及更高版本的 OpenSearch 版本设计的强大工具,该插件提供了强大的功能,同时通过原始 Elasticsearch 插件提供与 1.x 的兼容性。 此插件有助于在 OpenSearch 中创建和管理索引,自动管理模板并确保数据结构化以进行有效分析。 该插件支持各种配置选项,例如索引名称、身份验证、运行状况检查和值处理,使其可以根据不同的操作要求进行定制。 它的功能使其对于希望利用 OpenSearch 的强大功能进行指标存储和查询的组织至关重要。

配置

Kinesis


# Configuration for the AWS Kinesis input.
[[inputs.kinesis_consumer]]
  ## Amazon REGION of kinesis endpoint.
  region = "ap-southeast-2"

  ## Amazon Credentials
  ## Credentials are loaded in the following order
  ## 1) Web identity provider credentials via STS if role_arn and web_identity_token_file are specified
  ## 2) Assumed credentials via STS if role_arn is specified
  ## 3) explicit credentials from 'access_key' and 'secret_key'
  ## 4) shared profile from 'profile'
  ## 5) environment variables
  ## 6) shared credentials file
  ## 7) EC2 Instance Profile
  # access_key = ""
  # secret_key = ""
  # token = ""
  # role_arn = ""
  # web_identity_token_file = ""
  # role_session_name = ""
  # profile = ""
  # shared_credential_file = ""

  ## Endpoint to make request against, the correct endpoint is automatically
  ## determined and this option should only be set if you wish to override the
  ## default.
  ##   ex: endpoint_url = "http://localhost:8000"
  # endpoint_url = ""

  ## Kinesis StreamName must exist prior to starting telegraf.
  streamname = "StreamName"

  ## Shard iterator type (only 'TRIM_HORIZON' and 'LATEST' currently supported)
  # shard_iterator_type = "TRIM_HORIZON"

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

  ##
  ## The content encoding of the data from kinesis
  ## If you are processing a cloudwatch logs kinesis stream then set this to "gzip"
  ## as AWS compresses cloudwatch log data before it is sent to kinesis (aws
  ## also base64 encodes the zip byte data before pushing to the stream.  The base64 decoding
  ## is done automatically by the golang sdk, as data is read from kinesis)
  ##
  # content_encoding = "identity"

  ## Optional
  ## Configuration for a dynamodb checkpoint
  [inputs.kinesis_consumer.checkpoint_dynamodb]
    ## unique name for this consumer
    app_name = "default"
    table_name = "default"

OpenSearch

[[outputs.opensearch]]
  ## URLs
  ## The full HTTP endpoint URL for your OpenSearch instance. Multiple URLs can
  ## be specified as part of the same cluster, but only one URLs is used to
  ## write during each interval.
  urls = ["http://node1.os.example.com:9200"]

  ## Index Name
  ## Target index name for metrics (OpenSearch will create if it not exists).
  ## This is a Golang template (see https://pkg.go.dev/text/template)
  ## You can also specify
  ## metric name (`{{.Name}}`), tag value (`{{.Tag "tag_name"}}`), field value (`{{.Field "field_name"}}`)
  ## If the tag does not exist, the default tag value will be empty string "".
  ## the timestamp (`{{.Time.Format "xxxxxxxxx"}}`).
  ## For example: "telegraf-{{.Time.Format \"2006-01-02\"}}-{{.Tag \"host\"}}" would set it to telegraf-2023-07-27-HostName
  index_name = ""

  ## Timeout
  ## OpenSearch client timeout
  # timeout = "5s"

  ## Sniffer
  ## Set to true to ask OpenSearch a list of all cluster nodes,
  ## thus it is not necessary to list all nodes in the urls config option
  # enable_sniffer = false

  ## GZIP Compression
  ## Set to true to enable gzip compression
  # enable_gzip = false

  ## Health Check Interval
  ## Set the interval to check if the OpenSearch 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.
  # username = ""
  # password = ""
  ## HTTP bearer token authentication details
  # auth_bearer_token = ""

  ## Optional TLS Config
  ## Set to true/false to enforce TLS being enabled/disabled. If not set,
  ## enable TLS only if any of the other options are specified.
  # tls_enable =
  ## Trusted root certificates for server
  # tls_ca = "/path/to/cafile"
  ## Used for TLS client certificate authentication
  # tls_cert = "/path/to/certfile"
  ## Used for TLS client certificate authentication
  # tls_key = "/path/to/keyfile"
  ## Send the specified TLS server name via SNI
  # tls_server_name = "kubernetes.example.com"
  ## Use TLS but skip chain & host verification
  # insecure_skip_verify = false

  ## Template Config
  ## Manage templates
  ## 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

  ## Template Name
  ## The template name used for telegraf indexes
  # template_name = "telegraf"

  ## Overwrite Templates
  ## Set to true if you want telegraf to overwrite an existing template
  # overwrite_template = false

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

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

  ## Pipeline Name
  ## Additionally, you can specify a tag name using the notation (`{{.Tag "tag_name"}}`)
  ## which will be used as the pipeline name (e.g. "{{.Tag \"os_pipeline\"}}").
  ## 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.
  # default_pipeline = ""

输入和输出集成示例

Kinesis

  1. 使用 Kinesis 进行实时数据处理:此用例涉及将 Kinesis 插件与监控仪表板集成,以实时分析传入的数据指标。 例如,应用程序可以从多个服务使用日志并以可视化方式呈现它们,从而使运营团队能够快速识别趋势并对异常做出反应。

  2. 无服务器日志聚合:在无服务器架构中使用此插件,其中 Kinesis 流聚合来自各种微服务的日志。 该插件可以创建指标,帮助检测系统中的问题,通过第三方集成自动化警报流程,使团队能够最大限度地减少停机时间并提高可靠性。

  3. 基于流指标的动态扩展:实施一种解决方案,其中 Kinesis 插件使用的流指标可用于动态调整资源。 例如,如果处理的记录数激增,则可以触发相应的扩展操作来处理增加的负载,从而确保最佳资源分配和性能。

  4. 使用检查点的到 S3 的数据管道:创建一个强大的数据管道,其中 Kinesis 流数据通过 Telegraf Kinesis 插件处理,检查点存储在 DynamoDB 中。 这种方法可以确保数据一致性和可靠性,因为它管理已处理数据的状态,从而实现与下游数据湖或存储解决方案的无缝集成。

OpenSearch

  1. 时序数据的动态索引:利用 OpenSearch Telegraf 插件为时序指标动态创建索引,确保数据以有组织的方式存储,从而有利于基于时间的查询。 通过使用 Go 模板定义索引模式,用户可以利用该插件创建每日或每月索引,这可以大大简化数据的管理和长期检索,从而提高分析性能。

  2. 多租户应用程序的集中日志记录:在多租户应用程序中实施 OpenSearch 插件,其中每个租户的日志都发送到单独的索引。 这使得能够对每个租户进行有针对性的分析和监控,同时保持数据隔离。 通过利用索引名称模板功能,用户可以自动创建租户特定的索引,这不仅简化了流程,还提高了租户数据的安全性和可访问性。

  3. 与机器学习集成以进行异常检测:将 OpenSearch 插件与机器学习工具结合使用,以自动检测指标数据中的异常。 通过配置插件以将实时指标发送到 OpenSearch,用户可以在传入的数据流上应用机器学习模型,以识别异常值或不寻常的模式,从而促进主动监控和快速补救措施。

  4. 使用 OpenSearch 增强监控仪表板:使用从 OpenSearch 收集的指标创建实时仪表板,以提供对系统性能的洞察。 通过将指标馈送到 OpenSearch,组织可以利用 OpenSearch Dashboards 可视化关键绩效指标,使运营团队能够快速评估运行状况和性能,并做出数据驱动的决策。

反馈

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

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

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

查看入门方法

相关集成

HTTP 和 InfluxDB 集成

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

查看集成

Kafka 和 InfluxDB 集成

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

查看集成

Kinesis 和 InfluxDB 集成

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

查看集成