Kinesis 和 PostgreSQL 集成

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

info

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

50 亿+

Telegraf 下载量

#1

时间序列数据库
来源:DB Engines

10 亿+

InfluxDB 下载量

2,800+

贡献者

目录

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

收集、组织和处理大量高速数据。当您将任何数据视为时间序列数据时,它会更有价值。使用 InfluxDB,这是 #1 的时间序列平台,旨在与 Telegraf 一起扩展。

查看入门方法

输入和输出集成概述

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

Telegraf PostgreSQL 插件允许您有效地将指标写入 PostgreSQL 数据库,同时自动管理数据库模式。

集成详情

Kinesis

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

PostgreSQL

PostgreSQL 插件使用户能够将指标写入 PostgreSQL 数据库或兼容的数据库,通过自动更新缺失的列,为模式管理提供强大的支持。该插件旨在促进与监控解决方案的集成,允许用户有效地存储和管理时间序列数据。它为连接设置、并发和错误处理提供可配置的选项,并支持高级功能,例如用于标签和字段的 JSONB 存储、外键标记、模板化模式修改以及通过 pguint 扩展支持无符号整数数据类型。

配置

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"

PostgreSQL

# Publishes metrics to a postgresql database
[[outputs.postgresql]]
  ## Specify connection address via the standard libpq connection string:
  ##   host=... user=... password=... sslmode=... dbname=...
  ## Or a URL:
  ##   postgres://[user[:password]]@localhost[/dbname]?sslmode=[disable|verify-ca|verify-full]
  ## See https://postgresql.ac.cn/docs/current/libpq-connect.html#LIBPQ-CONNSTRING
  ##
  ## All connection parameters are optional. Environment vars are also supported.
  ## e.g. PGPASSWORD, PGHOST, PGUSER, PGDATABASE
  ## All supported vars can be found here:
  ##  https://postgresql.ac.cn/docs/current/libpq-envars.html
  ##
  ## Non-standard parameters:
  ##   pool_max_conns (default: 1) - Maximum size of connection pool for parallel (per-batch per-table) inserts.
  ##   pool_min_conns (default: 0) - Minimum size of connection pool.
  ##   pool_max_conn_lifetime (default: 0s) - Maximum age of a connection before closing.
  ##   pool_max_conn_idle_time (default: 0s) - Maximum idle time of a connection before closing.
  ##   pool_health_check_period (default: 0s) - Duration between health checks on idle connections.
  # connection = ""

  ## Postgres schema to use.
  # schema = "public"

  ## Store tags as foreign keys in the metrics table. Default is false.
  # tags_as_foreign_keys = false

  ## Suffix to append to table name (measurement name) for the foreign tag table.
  # tag_table_suffix = "_tag"

  ## Deny inserting metrics if the foreign tag can't be inserted.
  # foreign_tag_constraint = false

  ## Store all tags as a JSONB object in a single 'tags' column.
  # tags_as_jsonb = false

  ## Store all fields as a JSONB object in a single 'fields' column.
  # fields_as_jsonb = false

  ## Name of the timestamp column
  ## NOTE: Some tools (e.g. Grafana) require the default name so be careful!
  # timestamp_column_name = "time"

  ## Type of the timestamp column
  ## Currently, "timestamp without time zone" and "timestamp with time zone"
  ## are supported
  # timestamp_column_type = "timestamp without time zone"

  ## Templated statements to execute when creating a new table.
  # create_templates = [
  #   '''CREATE TABLE {{ .table }} ({{ .columns }})''',
  # ]

  ## Templated statements to execute when adding columns to a table.
  ## Set to an empty list to disable. Points containing tags for which there is no column will be skipped. Points
  ## containing fields for which there is no column will have the field omitted.
  # add_column_templates = [
  #   '''ALTER TABLE {{ .table }} ADD COLUMN IF NOT EXISTS {{ .columns|join ", ADD COLUMN IF NOT EXISTS " }}''',
  # ]

  ## Templated statements to execute when creating a new tag table.
  # tag_table_create_templates = [
  #   '''CREATE TABLE {{ .table }} ({{ .columns }}, PRIMARY KEY (tag_id))''',
  # ]

  ## Templated statements to execute when adding columns to a tag table.
  ## Set to an empty list to disable. Points containing tags for which there is no column will be skipped.
  # tag_table_add_column_templates = [
  #   '''ALTER TABLE {{ .table }} ADD COLUMN IF NOT EXISTS {{ .columns|join ", ADD COLUMN IF NOT EXISTS " }}''',
  # ]

  ## The postgres data type to use for storing unsigned 64-bit integer values (Postgres does not have a native
  ## unsigned 64-bit integer type).
  ## The value can be one of:
  ##   numeric - Uses the PostgreSQL "numeric" data type.
  ##   uint8 - Requires pguint extension (https://github.com/petere/pguint)
  # uint64_type = "numeric"

  ## When using pool_max_conns>1, and a temporary error occurs, the query is retried with an incremental backoff. This
  ## controls the maximum backoff duration.
  # retry_max_backoff = "15s"

  ## Approximate number of tag IDs to store in in-memory cache (when using tags_as_foreign_keys).
  ## This is an optimization to skip inserting known tag IDs.
  ## Each entry consumes approximately 34 bytes of memory.
  # tag_cache_size = 100000

  ## Enable & set the log level for the Postgres driver.
  # log_level = "warn" # trace, debug, info, warn, error, none

输入和输出集成示例

Kinesis

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

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

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

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

PostgreSQL

  1. 使用复杂查询进行实时分析:利用 PostgreSQL 插件将来自各种来源的指标存储在 PostgreSQL 数据库中,从而可以使用复杂查询进行实时分析。这种设置可以帮助数据科学家和分析师发现模式和趋势,因为他们可以在多个表之间操作关系数据,同时利用 PostgreSQL 强大的查询优化功能。具体来说,用户可以使用不同指标表上的 JOIN 操作创建复杂的报告,从而揭示通常隐藏在嵌入式系统中的见解。

  2. 与 TimescaleDB 集成以进行时间序列数据处理:在 TimescaleDB 实例中使用 PostgreSQL 插件,以高效地处理和分析时间序列数据。通过实施超表,用户可以在时间维度上实现更高的主题性能和分区。这种集成允许用户对大量时间序列数据运行分析查询,同时保留 PostgreSQL SQL 查询的全部功能,从而确保指标分析的可靠性和效率。

  3. 数据版本控制和历史分析:实施一种使用 PostgreSQL 插件来维护指标不同版本的策略。用户可以设置不可变的数据表结构,其中保留旧版本的表,从而方便进行历史分析。这种方法不仅可以深入了解数据演变,还有助于遵守数据保留策略,确保数据集的历史完整性保持不变。

  4. 用于不断发展的指标的动态模式管理:使用插件的模板功能来创建动态变化的模式,以响应指标变化。此用例允许组织在指标演变时调整其数据结构,添加必要的字段并确保遵守数据完整性策略。通过利用模板化的 SQL 命令,用户可以在无需人工干预的情况下扩展其数据库,从而促进敏捷的数据管理实践。

反馈

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

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

收集、组织和处理大量高速数据。当您将任何数据视为时间序列数据时,它会更有价值。使用 InfluxDB,这是 #1 的时间序列平台,旨在与 Telegraf 一起扩展。

查看入门方法

相关集成

HTTP 和 InfluxDB 集成

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

查看集成

Kafka 和 InfluxDB 集成

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

查看集成

Kinesis 和 InfluxDB 集成

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

查看集成