文档文档

Kapacitor

FromNode

from 节点选择流经 StreamNode 的数据子集。stream 节点允许您选择要处理的流的哪一部分。

示例

stream
  |from()
    .database('mydb')
    .retentionPolicy('myrp')
    .measurement('mymeasurement')
    .where(lambda: "host" =~ /logger\d+/)
  |window()
  ...

上面的示例仅选择来自数据库 mydb 和保留策略 myrp 以及指标 mymeasurement 的数据点，其中标签 host 匹配正则表达式 logger\d+。

构造函数

链式方法描述
from ( )创建一个新的流节点，可以使用 Database、RetentionPolicy、Measurement 和 Where 属性进一步过滤。可以多次调用 From 以创建数据流的多个独立分支。

属性方法

设置器描述
database ( value string)数据库名称。如果为空，将使用任何数据库。
groupBy ( tag ...interface{})按一组标签对数据进行分组。
groupByMeasurement ( )如果设置，将在组 ID 中包含指标名称。以及任何其他分组维度。
measurement ( value string)指标名称。如果为空，将使用任何指标。
quiet ( )禁止来自此节点的所有错误日志记录事件。
retentionPolicy ( value string)保留策略名称。如果为空，将使用任何保留策略。
round ( value time.Duration)用于舍入时间戳的可选持续时间。有助于确保数据点落在特定边界上。示例：stream
truncate ( value time.Duration)用于截断时间戳的可选持续时间。有助于确保数据点落在特定边界上。示例：stream
where ( lambda ast.LambdaNode)使用给定的表达式过滤当前流。此表达式是 Kapacitor 表达式。Kapacitor 表达式是 InfluxQL WHERE 表达式的超集。有关更多信息，请参阅 expression 文档。

链式方法

Alert, Barrier, Bottom, ChangeDetect, Combine, Count, CumulativeSum, Deadman, Default, Delete, Derivative, Difference, Distinct, Ec2Autoscale, Elapsed, Eval, First, Flatten, From, 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, Union, Window

属性

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

数据库

数据库名称。如果为空，将使用任何数据库。

from.database(value string)

GroupBy

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

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

  stream
      |from()
          .groupBy(*)

from.groupBy(tag ...interface{})

GroupByMeasurement

如果设置，将在组 ID 中包含指标名称。以及任何其他分组维度。

示例

 stream
      |from()
          .database('mydb')
          .groupByMeasurement()
          .groupBy('host')

上面的示例选择数据库“mydb”中的所有指标，然后每个点按 host 标签和指标名称分组。因此，将指标保留在自己的组中。

from.groupByMeasurement()

Measurement

指标名称。如果为空，将使用任何指标。

from.measurement(value string)

Quiet

禁止来自此节点的所有错误日志记录事件。

from.quiet()

RetentionPolicy

保留策略名称。如果为空，将使用任何保留策略。

from.retentionPolicy(value string)

Round

用于舍入时间戳的可选持续时间。有助于确保数据点落在特定边界上。示例

    stream
       |from()
           .measurement('mydata')
           .round(1s)

所有传入数据都将舍入到最接近的 1 秒边界。

from.round(value time.Duration)

Truncate

用于截断时间戳的可选持续时间。有助于确保数据点落在特定边界上。示例

    stream
       |from()
           .measurement('mydata')
           .truncate(1s)

所有传入数据都将被截断为 1 秒分辨率。

from.truncate(value time.Duration)

Where

使用给定的表达式过滤当前流。此表达式是 Kapacitor 表达式。Kapacitor 表达式是 InfluxQL WHERE 表达式的超集。有关更多信息，请参阅 expression 文档。

多次调用 Where 方法会将每个表达式 AND 在一起。

示例

    stream
       |from()
          .where(lambda: condition1)
          .where(lambda: condition2)

上面等效于此示例

    stream
       |from()
          .where(lambda: condition1 AND condition2)

注意：如果要使用多个不同的流，请务必始终使用 |from。

示例

  var data = stream
      |from()
          .measurement('cpu')
  var total = data
      .where(lambda: "cpu" == 'cpu-total')
  var others = data
      .where(lambda: "cpu" != 'cpu-total')

上面的示例等效于下面的示例，这显然不是预期的结果。

示例

  var data = stream
      |from()
          .measurement('cpu')
          .where(lambda: "cpu" == 'cpu-total' AND "cpu" != 'cpu-total')
  var total = data
  var others = total

下面的示例将创建两个不同的流，每个流选择原始流的不同子集。

示例

  var data = stream
      |from()
          .measurement('cpu')
  var total = stream
      |from()
          .measurement('cpu')
          .where(lambda: "cpu" == 'cpu-total')
  var others = stream
      |from()
          .measurement('cpu')
          .where(lambda: "cpu" != 'cpu-total')

如果为空，则所有数据点都被视为匹配。

from.where(lambda ast.LambdaNode)

链式方法

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

Alert

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

from|alert()

返回：AlertNode

Barrier

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

每个周期持续时间将发出一个 BarrierMessage。

from|barrier()

返回：BarrierNode

Bottom

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

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

返回：InfluxQLNode

ChangeDetect

创建一个新节点，仅当新点与上一个点不同时才发出新点。

from|changeDetect(field string)

返回：ChangeDetectNode

Combine

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

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

返回：CombineNode

Count

计算点的数量。

from|count(field string)

返回：InfluxQLNode

CumulativeSum

计算接收到的每个点的累积总和。为收集的每个点发出一个点。

from|cumulativeSum(field string)

返回：InfluxQLNode

Deadman

用于在低吞吐量（又名“deadman’s switch”）上创建警报的辅助函数。

  • 阈值：如果吞吐量在 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 表达式，以进一步约束何时触发“deadman’s switch”。示例

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

from|deadman(threshold float64, interval time.Duration, expr ...ast.LambdaNode)

返回：AlertNode

Default

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

from|default()

返回：DefaultNode

Delete

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

from|delete()

返回：DeleteNode

Derivative

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

from|derivative(field string)

返回：DerivativeNode

Difference

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

from|difference(field string)

返回：InfluxQLNode

Distinct

生成仅包含不同点的批处理。

from|distinct(field string)

返回：InfluxQLNode

Ec2Autoscale

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

from|ec2Autoscale()

返回：Ec2AutoscaleNode

Elapsed

计算点之间经过的时间。

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

返回：InfluxQLNode

Eval

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

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

返回：EvalNode

First

选择第一个点。

from|first(field string)

返回：InfluxQLNode

Flatten

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

from|flatten()

返回：FlattenNode

From

创建一个新的流节点，可以使用 Database、RetentionPolicy、Measurement 和 Where 属性进一步过滤。可以多次调用 From 以创建数据流的多个独立分支。

示例

    // Select the 'cpu' measurement from just the database 'mydb'
    // and retention policy 'myrp'.
    var cpu = stream
        |from()
            .database('mydb')
            .retentionPolicy('myrp')
            .measurement('cpu')
    // Select the 'load' measurement from any database and retention policy.
    var load = stream
        |from()
            .measurement('load')
    // Join cpu and load streams and do further processing.
    cpu
        |join(load)
            .as('cpu', 'load')
        ...

from|from()

返回：FromNode

HoltWinters

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

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

返回：InfluxQLNode

HoltWintersWithFit

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

from|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 请求数据。

from|httpOut(endpoint string)

返回：HTTPOutNode

HttpPost

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

from|httpPost(url ...string)

返回：HTTPPostNode

InfluxDBOut

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

from|influxDBOut()

返回：InfluxDBOutNode

Join

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

from|join(others ...Node)

返回：JoinNode

K8sAutoscale

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

from|k8sAutoscale()

返回：K8sAutoscaleNode

KapacitorLoopback

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

from|kapacitorLoopback()

返回：KapacitorLoopbackNode

Last

选择最后一个点。

from|last(field string)

返回：InfluxQLNode

Log

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

from|log()

返回：LogNode

Max

选择最大点。

from|max(field string)

返回：InfluxQLNode

Mean

计算数据的平均值。

from|mean(field string)

返回：InfluxQLNode

Median

计算数据的中位数。

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

from|median(field string)

返回：InfluxQLNode

Min

选择最小点。

from|min(field string)

返回：InfluxQLNode

Mode

计算数据的众数。

from|mode(field string)

返回：InfluxQLNode

MovingAverage

计算最后 window 个点的移动平均值。在窗口填满之前，不会发出任何点。

from|movingAverage(field string, window int64)

返回：InfluxQLNode

Percentile

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

from|percentile(field string, percentile float64)

返回：InfluxQLNode

Sample

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

将每计数或指定持续时间发出一个点。

from|sample(rate interface{})

返回：SampleNode

Shift

创建一个新节点，及时移动传入的点或批处理。

from|shift(shift time.Duration)

返回：ShiftNode

Sideload

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

from|sideload()

返回：SideloadNode

Spread

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

from|spread(field string)

返回：InfluxQLNode

StateCount

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

from|stateCount(expression ast.LambdaNode)

返回：StateCountNode

StateDuration

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

from|stateDuration(expression ast.LambdaNode)

返回：StateDurationNode

Stats

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

from|stats(interval time.Duration)

返回：StatsNode

Stddev

计算标准偏差。

from|stddev(field string)

返回：InfluxQLNode

Sum

计算所有值的总和。

from|sum(field string)

返回：InfluxQLNode

SwarmAutoscale

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

from|swarmAutoscale()

返回：SwarmAutoscaleNode

Top

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

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

返回：InfluxQLNode

Union

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

from|union(node ...Node)

返回：UnionNode

Window

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

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

from|window()

返回：WindowNode

此页内容对您有帮助吗？

感谢您的反馈！

© 2025 InfluxData, Inc.

您在哪里运行 InfluxDB？

选择您的 InfluxDB Cloud 区域和集群 或您的 InfluxDB OSS URL，我们将为您自定义代码示例。确定您的 InfluxDB Cloud 集群。

  • InfluxDB Cloud
  • InfluxDB OSS 或 Enterprise
AWS

  • 美西（俄勒冈州）

GCP
Azure
Default
自定义

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

感谢您的反馈！

请告诉我们如何做得更好

谢谢！

不用了，谢谢

Flux 的未来

Flux 即将进入维护模式。您可以继续像现在一样使用它，而无需对您的代码进行任何更改。

阅读更多

InfluxDB 3 开源现已公开发布 Alpha 版本

InfluxDB 3 开源版本现在可用于 Alpha 测试，根据 MIT 或 Apache 2 许可获得许可。

我们正在发布两个产品作为 Alpha 版本的一部分。

InfluxDB 3 Core 是我们新的开源产品。它是一个用于时间序列和事件数据的最新数据引擎。InfluxDB 3 Enterprise 是一个商业版本，它建立在 Core 的基础上，增加了历史查询功能、读取副本、高可用性、可伸缩性和细粒度的安全性。

有关如何开始使用的更多信息，请查看