OPC UA 和 TimescaleDB 集成

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

info

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

50 亿+

Telegraf 下载量

#1

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

10 亿+

InfluxDB 下载量

2,800+

贡献者

目录

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

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

查看入门方法

输入和输出集成概述

OPC UA 插件提供了一个从 OPC UA 服务器设备检索数据的接口,从而促进有效的数据收集和监控。

此输出插件为将 Telegraf 收集的指标直接路由到 TimescaleDB 提供了可靠高效的机制。通过利用 PostgreSQL 的强大生态系统以及 TimescaleDB 的时间序列优化,它支持高性能数据摄取和高级查询功能。

集成详情

OPC UA

OPC UA 插件从使用 OPC UA 协议通信的设备检索数据,使您可以收集和监控来自 OPC UA 服务器的数据。

TimescaleDB

TimescaleDB 是一个开源时间序列数据库,作为 PostgreSQL 的扩展构建,旨在高效处理大规模、面向时间的数据。TimescaleDB 于 2017 年推出,是为了响应对强大、可扩展的解决方案日益增长的需求而诞生的,该解决方案可以管理海量数据、高插入速率和复杂查询。通过利用 PostgreSQL 熟悉的 SQL 接口,并使用专门的时间序列功能对其进行增强,TimescaleDB 迅速在希望将时间序列功能集成到现有关系数据库中的开发人员中流行起来。它的混合方法允许用户受益于 PostgreSQL 的灵活性、可靠性和生态系统,同时为时间序列数据提供优化的性能。

该数据库在需要快速摄取数据点并结合对历史周期进行复杂分析查询的环境中尤其有效。TimescaleDB 具有许多创新功能,例如将数据透明地分区为可管理块的超表和内置的连续聚合。这些功能显着提高了查询速度和资源效率。

配置

OPC UA


[[inputs.opcua]]
  ## Metric name
  # name = "opcua"
  #
  ## OPC UA Endpoint URL
  # endpoint = "opc.tcp://localhost:4840"
  #
  ## Maximum time allowed to establish a connect to the endpoint.
  # connect_timeout = "10s"
  #
  ## Maximum time allowed for a request over the established connection.
  # request_timeout = "5s"

  # Maximum time that a session shall remain open without activity.
  # session_timeout = "20m"
  #
  ## Security policy, one of "None", "Basic128Rsa15", "Basic256",
  ## "Basic256Sha256", or "auto"
  # security_policy = "auto"
  #
  ## Security mode, one of "None", "Sign", "SignAndEncrypt", or "auto"
  # security_mode = "auto"
  #
  ## Path to cert.pem. Required when security mode or policy isn't "None".
  ## If cert path is not supplied, self-signed cert and key will be generated.
  # certificate = "/etc/telegraf/cert.pem"
  #
  ## Path to private key.pem. Required when security mode or policy isn't "None".
  ## If key path is not supplied, self-signed cert and key will be generated.
  # private_key = "/etc/telegraf/key.pem"
  #
  ## Authentication Method, one of "Certificate", "UserName", or "Anonymous".  To
  ## authenticate using a specific ID, select 'Certificate' or 'UserName'
  # auth_method = "Anonymous"
  #
  ## Username. Required for auth_method = "UserName"
  # username = ""
  #
  ## Password. Required for auth_method = "UserName"
  # password = ""
  #
  ## Option to select the metric timestamp to use. Valid options are:
  ##     "gather" -- uses the time of receiving the data in telegraf
  ##     "server" -- uses the timestamp provided by the server
  ##     "source" -- uses the timestamp provided by the source
  # timestamp = "gather"
  #
  ## Client trace messages
  ## When set to true, and debug mode enabled in the agent settings, the OPCUA
  ## client's messages are included in telegraf logs. These messages are very
  ## noisey, but essential for debugging issues.
  # client_trace = false
  #
  ## Include additional Fields in each metric
  ## Available options are:
  ##   DataType -- OPC-UA Data Type (string)
  # optional_fields = []
  #
  ## Node ID configuration
  ## name              - field name to use in the output
  ## namespace         - OPC UA namespace of the node (integer value 0 thru 3)
  ## identifier_type   - OPC UA ID type (s=string, i=numeric, g=guid, b=opaque)
  ## identifier        - OPC UA ID (tag as shown in opcua browser)
  ## tags              - extra tags to be added to the output metric (optional); deprecated in 1.25.0; use default_tags
  ## default_tags      - extra tags to be added to the output metric (optional)
  ##
  ## Use either the inline notation or the bracketed notation, not both.
  #
  ## Inline notation (default_tags not supported yet)
  # nodes = [
  #   {name="", namespace="", identifier_type="", identifier="", tags=[["tag1", "value1"], ["tag2", "value2"]},
  #   {name="", namespace="", identifier_type="", identifier=""},
  # ]
  #
  ## Bracketed notation
  # [[inputs.opcua.nodes]]
  #   name = "node1"
  #   namespace = ""
  #   identifier_type = ""
  #   identifier = ""
  #   default_tags = { tag1 = "value1", tag2 = "value2" }
  #
  # [[inputs.opcua.nodes]]
  #   name = "node2"
  #   namespace = ""
  #   identifier_type = ""
  #   identifier = ""
  #
  ## Node Group
  ## Sets defaults so they aren't required in every node.
  ## Default values can be set for:
  ## * Metric name
  ## * OPC UA namespace
  ## * Identifier
  ## * Default tags
  ##
  ## Multiple node groups are allowed
  #[[inputs.opcua.group]]
  ## Group Metric name. Overrides the top level name.  If unset, the
  ## top level name is used.
  # name =
  #
  ## Group default namespace. If a node in the group doesn't set its
  ## namespace, this is used.
  # namespace =
  #
  ## Group default identifier type. If a node in the group doesn't set its
  ## namespace, this is used.
  # identifier_type =
  #
  ## Default tags that are applied to every node in this group. Can be
  ## overwritten in a node by setting a different value for the tag name.
  ##   example: default_tags = { tag1 = "value1" }
  # default_tags = {}
  #
  ## Node ID Configuration.  Array of nodes with the same settings as above.
  ## Use either the inline notation or the bracketed notation, not both.
  #
  ## Inline notation (default_tags not supported yet)
  # nodes = [
  #  {name="node1", namespace="", identifier_type="", identifier=""},
  #  {name="node2", namespace="", identifier_type="", identifier=""},
  #]
  #
  ## Bracketed notation
  # [[inputs.opcua.group.nodes]]
  #   name = "node1"
  #   namespace = ""
  #   identifier_type = ""
  #   identifier = ""
  #   default_tags = { tag1 = "override1", tag2 = "value2" }
  #
  # [[inputs.opcua.group.nodes]]
  #   name = "node2"
  #   namespace = ""
  #   identifier_type = ""
  #   identifier = ""

  ## Enable workarounds required by some devices to work correctly
  # [inputs.opcua.workarounds]
    ## Set additional valid status codes, StatusOK (0x0) is always considered valid
  # additional_valid_status_codes = ["0xC0"]

  # [inputs.opcua.request_workarounds]
    ## Use unregistered reads instead of registered reads
  # use_unregistered_reads = false

TimescaleDB

# Publishes metrics to a TimescaleDB 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 connection age 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 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

  ## Cut column names at the given length to not exceed PostgreSQL's
  ## 'identifier length' limit (default: no limit)
  ## (see https://postgresql.ac.cn/docs/current/limits.html)
  ## Be careful to not create duplicate column names!
  # column_name_length_limit = 0

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

输入和输出集成示例

OPC UA

  1. 基本配置:使用您的 OPC UA 服务器端点和所需的指标设置插件。这允许 Telegraf 开始从配置的节点收集指标。

  2. 节点 ID 设置:使用配置指定特定节点,例如温度传感器,以实时监控它们的值。例如,配置节点 ns=3;s=Temperature 以直接收集温度数据。

  3. 组配置:通过将多个节点分组在一个配置下,简化对它们的监控 - 这将为该组中的所有节点设置默认值,从而减少设置中的冗余。

TimescaleDB

  1. 实时物联网数据摄取:使用插件实时收集和存储来自数千个物联网设备的传感器数据。此设置有助于即时分析,帮助组织监控运营效率并快速响应不断变化的条件。

  2. 云应用程序性能监控:利用插件将来自分布式云应用程序的详细性能指标馈送到 TimescaleDB 中。这种集成支持实时仪表板和警报,使团队能够快速识别和缓解性能瓶颈。

  3. 历史数据分析和报告:实施一个系统,将长期指标存储在 TimescaleDB 中,以进行全面的历史分析。这种方法使企业能够执行趋势分析、生成详细报告并根据存档的时间序列数据做出数据驱动的决策。

  4. 自适应警报和异常检测:将插件与自动异常检测工作流程集成。通过将指标持续流式传输到 TimescaleDB,机器学习模型可以分析数据模式,并在发生异常时触发警报,从而提高系统可靠性和主动维护能力。

反馈

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

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

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

查看入门方法

相关集成

HTTP 和 InfluxDB 集成

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

查看集成

Kafka 和 InfluxDB 集成

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

查看集成

Kinesis 和 InfluxDB 集成

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

查看集成