文档文档

Kapacitor

SideloadNode

sideload 节点根据来自各种源的层次化数据，向数据点添加字段和标签。

SideloadNode 示例

以下示例展示了如何根据从层次化源加载的值，向每个数据点添加一个字段 cpu_threshold 和一个标签 foo。.order() 属性中的模板列表使用数据点的标签进行评估。然后按顺序检查文件路径以获取指定的键，并使用找到的第一个值。

|sideload()
  .source('file:///path/to/dir')
  .order('host/{{.host}}.yml', 'hostgroup/{{.hostgroup}}.yml')
  .field('cpu_threshold', 0.0)
  .tag('foo', 'unknown')

构造函数

链式方法描述
sideload ( )创建一个可以从外部源加载数据的节点

属性方法

Setter 方法描述
field ( f string, v interface{})Field 是要从源加载的字段的名称及其默认值。加载的类型必须与默认值的类型匹配。否则，将记录一个错误并使用默认值。
order ( order ...string)Order 是一个指示层次化顺序的路径列表。路径相对于源，并且可以包含模板标记，如 {{.tagname}}，这些标记将被替换为数据点的标签值。然后按顺序搜索路径以获取键，并使用找到的第一个值。这允许基于标签的层次结构来覆盖值。
quiet ( )抑制此节点的所有错误日志事件。
source ( value string)数据的源，例如，file:// 或 http://
tag ( t string, v string)Tag 是要从源加载的标签的名称及其默认值。加载的值必须是字符串，否则将记录一个错误并使用默认值。

链式方法

Alert, Barrier, Bottom, ChangeDetect, Combine, Count, CumulativeSum, Deadman, Default, Delete, Derivative, Difference, Distinct, Ec2Autoscale, Elapsed, Eval, First, Flatten, GroupBy, HoltWinters, HoltWintersWithFit, HttpOut, HttpPost, InfluxDBOut, Join, K8sAutoscale, KapacitorLoopback, Last, Log, Max, Mean, Median, Min, Mode, MovingAverage, Percentile, Sample, Shift, Sideload, Spread, StateCount, StateDuration, Stats, Stddev, Sum, SwarmAutoscale, Top, Trickle, Union, Where, Window

属性

属性方法会修改调用节点的状态。它们不会向管道添加新节点，并且始终返回对调用节点的引用。属性方法使用 . 运算符标记。

字段

Field 是要从源加载的字段的名称及其默认值。加载的类型必须与默认值的类型匹配。否则，将记录一个错误并使用默认值。

sideload.field(f string, v interface{})

Order

Order 是一个指示层次化顺序的路径列表。路径相对于源，并且可以包含模板标记，如 {{.tagname}}，这些标记将被替换为数据点的标签值。然后按顺序搜索路径以获取键，并使用找到的第一个值。这允许基于标签的层次结构来覆盖值。

sideload.order(order ...string)

Quiet

抑制此节点的所有错误日志事件。

sideload.quiet()

来源

定义数据源。支持以下源

  • 文件 URI (file://)
  • URL (http://)
  • 端点名称。使用纯字符串（不带 file:// 或 http://）以确保源在 Kapacitor 配置的 [[httpost]] 部分被解释为端点。

源示例

文件源
|sideload()
  .source('file:///path/to/dir')
  .order('host/{{.host}}.yml', 'hostgroup/{{.hostgroup}}.yml')
  .field('cpu_threshold', 0.0)
  .tag('foo', 'unknown')
URL 源
|sideload()
  .source('https://:5000/threshold/')
  .order('host/{{.host}}.yml', 'hostgroup/{{.hostgroup}}.yml')
  .field('cpu_threshold', 0.0)
  .tag('foo', 'unknown')
端点源
|sideload()
  .source('host1')
  .order('host/{{.host}}.yml', 'hostgroup/{{.hostgroup}}.yml')
  .field('cpu_threshold', 0.0)
  .tag('foo', 'unknown')

Kapacitor 在任务启用时使用 HTTP GET 请求加载一次 URL 或 端点 源，然后在后续调用 /sideload/reload 端点时进行加载。

HTTP 源端点应返回一个 JSON 对象，其中每个属性是 order 语句中指定的键名，其值是一个包含键值对集合的对象。

{
    "host1": {
        "cpu_threshold": 99.9,
        "disable": "false"
    },
    "host2": {
        "cpu_threshold": 85,
        "disable": "true"
    },
    "host3": {
        "cpu_threshold": 56,
        "disable": "false"
    }
}

标签

Tag 是要从源加载的标签的名称及其默认值。加载的值必须是字符串，否则将记录一个错误并使用默认值。

sideload.tag(t string, v string)

链式方法

链式方法在管道中创建一个新的节点作为调用节点的子节点。它们不会修改调用节点。链式方法使用 | 运算符标记。

警报（Alert）

创建一个警报节点，可以触发警报。

sideload|alert()

返回：AlertNode

Barrier

创建一个新的 Barrier 节点，该节点会定期发出 BarrierMessage。

每个 period duration 将发出一个 BarrierMessage。

sideload|barrier()

返回：BarrierNode

Bottom

选择 field 的底部 num 个点，并按任何额外的标签或字段排序。

sideload|bottom(num int64, field string, fieldsAndTags ...string)

返回：InfluxQLNode

ChangeDetect

创建一个新节点，该节点仅在点与前一个点不同时才发出新点。

sideload|changeDetect(field string)

返回：ChangeDetectNode

Combine

将此节点与自身合并。数据按时间戳合并。

sideload|combine(expressions ...ast.LambdaNode)

返回：CombineNode

Count

计算点的数量。

sideload|count(field string)

返回：InfluxQLNode

CumulativeSum

计算接收到的每个点的累积总和。对于收集到的每个点都会发出一个点。

sideload|cumulativeSum(field string)

返回：InfluxQLNode

死人开关（Deadman）

用于创建低吞吐量警报（即死人开关）的辅助函数。

  • 阈值：如果吞吐量低于每 points/interval 的点数，则触发警报。
  • 间隔：检查吞吐量的频率。
  • 表达式：可选的表达式列表，也用于评估。对于按时警报很有用。

示例

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    data
        |deadman(100.0, 10s)
    //Do normal processing of data
    data...

以上等同于以下示例

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    data
        |stats(10s)
            .align()
        |derivative('emitted')
            .unit(10s)
            .nonNegative()
        |alert()
            .id('node \'stream0\' in task \'{{ .TaskName }}\'')
            .message('{{ .ID }} is {{ if eq .Level "OK" }}alive{{ else }}dead{{ end }}: {{ index .Fields "emitted" | printf "%0.3f" }} points/10s.')
            .crit(lambda: "emitted" <= 100.0)
    //Do normal processing of data
    data...

id 和 message 警报属性可以通过 'deadman' 配置部分全局配置。

由于 AlertNode 是最后一个组件，因此可以像往常一样对其进行进一步修改。示例

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    data
        |deadman(100.0, 10s)
            .slack()
            .channel('#dead_tasks')
    //Do normal processing of data
    data...

您可以指定额外的 lambda 表达式来进一步限制死人开关的触发时间。示例

    var data = stream
        |from()...
    // Trigger critical alert if the throughput drops below 100 points per 10s and checked every 10s.
    // Only trigger the alert if the time of day is between 8am-5pm.
    data
        |deadman(100.0, 10s, lambda: hour("time") >= 8 AND hour("time") <= 17)
    //Do normal processing of data
    data...
sideload|deadman(threshold float64, interval time.Duration, expr ...ast.LambdaNode)

返回：AlertNode

默认

创建一个可以为缺失的标签或字段设置默认值的节点。

sideload|default()

返回：DefaultNode

删除

创建一个可以删除标签或字段的节点。

sideload|delete()

返回：DeleteNode

导数

创建一个新节点，该节点计算相邻点的导数。

sideload|derivative(field string)

返回：DerivativeNode

Difference

计算点之间的差异，与经过的时间无关。

sideload|difference(field string)

返回：InfluxQLNode

Distinct

生成仅包含不同点的批次。

sideload|distinct(field string)

返回：InfluxQLNode

Ec2Autoscale

创建一个可以为 ec2 自动伸缩组触发自动伸缩事件的节点。

sideload|ec2Autoscale()

返回：Ec2AutoscaleNode

Elapsed

计算点之间经过的时间。

sideload|elapsed(field string, unit time.Duration)

返回：InfluxQLNode

Eval

创建一个 eval 节点，该节点将评估给定的转换函数到每个数据点。可以提供表达式列表，并将按给定顺序评估它们。结果可供后续表达式使用。

sideload|eval(expressions ...ast.LambdaNode)

返回：EvalNode

First

选择第一个点。

sideload|first(field string)

返回：InfluxQLNode

Flatten

将具有相似时间戳的点展平为单个点。

sideload|flatten()

返回：FlattenNode

GroupBy

按标签集对数据进行分组。

可以传递字面值 * 来按所有维度分组。示例

    |groupBy(*)
sideload|groupBy(tag ...interface{})

返回：GroupByNode

HoltWinters

计算 Holt-Winters (/influxdb/v1/query_language/functions/#holt-winters) 数据集的预测。

sideload|holtWinters(field string, h int64, m int64, interval time.Duration)

返回：InfluxQLNode

HoltWintersWithFit

计算 Holt-Winters (/influxdb/v1/query_language/functions/#holt-winters) 数据集的预测。此方法除了预测数据外，还会输出用于拟合数据的所有点。

sideload|holtWintersWithFit(field string, h int64, m int64, interval time.Duration)

返回：InfluxQLNode

HttpOut

创建一个 HTTP 输出节点，该节点缓存它接收到的最新数据。缓存的数据可在给定端点处获得。端点是运行任务的 API 端点的相对路径。例如，如果任务端点位于 /kapacitor/v1/tasks/<task_id> 且端点为 top10，则可以从 /kapacitor/v1/tasks/<task_id>/top10 请求数据。

sideload|httpOut(endpoint string)

返回：HTTPOutNode

HttpPost

创建一个 HTTP Post 节点，该节点将接收到的数据 POST 到提供的 HTTP 端点。HttpPost 期望 0 个或 1 个参数。如果提供 0 个参数，则必须指定一个 endpoint 属性方法。

sideload|httpPost(url ...string)

返回：HTTPPostNode

InfluxDBOut

创建一个 influxdb 输出节点，该节点会将传入的数据存储到 InfluxDB 中。

sideload|influxDBOut()

返回：InfluxDBOutNode

Join

将此节点与其他节点连接。数据按时间戳连接。

sideload|join(others ...Node)

返回：JoinNode

K8sAutoscale

创建一个可以为 kubernetes 集群触发自动伸缩事件的节点。

sideload|k8sAutoscale()

返回：K8sAutoscaleNode

KapacitorLoopback

创建一个 kapacitor 回送节点，该节点会将数据作为流重新发送到 Kapacitor 中。

sideload|kapacitorLoopback()

返回：KapacitorLoopbackNode

Last

选择最后一个点。

sideload|last(field string)

返回：InfluxQLNode

Log

创建一个记录其接收的所有数据的节点。

sideload|log()

返回：LogNode

Max

选择最大点。

sideload|max(field string)

返回：InfluxQLNode

平均值

计算数据的平均值。

sideload|mean(field string)

返回：InfluxQLNode

Median

计算数据的中位数。

注意：此方法不是选择器。如果需要中位数点，请使用 .percentile(field, 50.0)。

sideload|median(field string)

返回：InfluxQLNode

Min

选择最小点。

sideload|min(field string)

返回：InfluxQLNode

Mode

计算数据的众数。

sideload|mode(field string)

返回：InfluxQLNode

MovingAverage

计算最后 window 个点的移动平均值。直到窗口填满才会发出点。

sideload|movingAverage(field string, window int64)

返回：InfluxQLNode

Percentile

选择给定百分位数的点。这是一个选择器函数，不会在点之间进行插值。

sideload|percentile(field string, percentile float64)

返回：InfluxQLNode

Sample

创建一个新节点，该节点对传入的点或批次进行采样。

将发出一个点，每隔指定的 count 或 duration。

sideload|sample(rate interface{})

返回：SampleNode

Shift

创建一个新节点，该节点将传入的点或批次在时间上进行移位。

sideload|shift(shift time.Duration)

返回：ShiftNode

Sideload

创建一个可以从外部源加载数据的节点。

sideload|sideload()

返回：SideloadNode

Spread

计算 min 和 max 点之间的差值。

sideload|spread(field string)

返回：InfluxQLNode

StateCount

创建一个节点，该节点跟踪给定状态下的连续点数。

sideload|stateCount(expression ast.LambdaNode)

返回：StateCountNode

StateDuration

创建一个节点，该节点跟踪给定状态下的持续时间。

sideload|stateDuration(expression ast.LambdaNode)

返回：StateDurationNode

Stats

创建一个新的数据流，其中包含节点的内部统计信息。interval 表示基于实际时间发出统计信息的频率。这意味着 interval 时间与源节点接收的数据点的时间无关。

sideload|stats(interval time.Duration)

返回：StatsNode

Stddev

计算标准偏差。

sideload|stddev(field string)

返回：InfluxQLNode

Sum

计算所有值的总和。

sideload|sum(field string)

返回：InfluxQLNode

SwarmAutoscale

创建一个可以为 Docker swarm 集群触发自动伸缩事件的节点。

sideload|swarmAutoscale()

返回：SwarmAutoscaleNode

Top

选择 field 的顶部 num 个点，并按任何额外的标签或字段排序。

sideload|top(num int64, field string, fieldsAndTags ...string)

返回：InfluxQLNode

Trickle

创建一个新节点，该节点将批处理数据转换为流数据。

sideload|trickle()

返回：TrickleNode

Union

执行此节点与所有其他给定节点的并集。

sideload|union(node ...Node)

返回：UnionNode

Where

创建一个新节点，该节点根据给定表达式过滤数据流。

sideload|where(expression ast.LambdaNode)

返回：WhereNode

窗口

创建一个新节点，该节点按时间对流进行窗口化。

注意：Window 只能应用于流边缘。

sideload|window()

返回：WindowNode

此页面是否有帮助？

感谢您的反馈！

© 2025 InfluxData, Inc.

您在哪里运行 InfluxDB？

选择您的 **InfluxDB Cloud 区域和集群** 或您的 **InfluxDB OSS URL**，我们将为您定制代码示例。识别您的 InfluxDB Cloud 集群。

  • InfluxDB Cloud
  • InfluxDB OSS 或 Enterprise
AWS

  • 美国西部（俄勒冈州）

GCP
Azure
默认
自定义

有关更多信息，请参阅 InfluxDB Cloud 区域 或 InfluxDB OSS URL。

感谢您的反馈！

让我们知道我们可以做得更好

谢谢！

不用了，谢谢

InfluxDB 3.8 新特性

InfluxDB 3.8 和 InfluxDB 3 Explorer 1.6 的主要增强功能。

查看博客文章

InfluxDB 3.8 现已适用于 Core 和 Enterprise 版本，同时发布了 InfluxDB 3 Explorer UI 的 1.6 版本。本次发布着重于操作成熟度，以及如何更轻松地部署、管理和可靠地运行 InfluxDB。

更多信息，请查看

InfluxDB Docker 的 latest 标签将指向 InfluxDB 3 Core

在 **2026 年 2 月 3 日**，InfluxDB Docker 镜像的 latest 标签将指向 InfluxDB 3 Core。为避免意外升级，请在您的 Docker 部署中使用特定的版本标签。

如果使用 Docker 来安装和运行 InfluxDB，latest 标签将指向 InfluxDB 3 Core。为避免意外升级，请在您的 Docker 部署中使用特定的版本标签。例如，如果使用 Docker 运行 InfluxDB v2，请将 latest 版本标签替换为 Docker pull 命令中的特定版本标签 — 例如

docker pull influxdb:2