文档文档

Kapacitor

QueryNode

query 节点定义了用于处理批处理数据的数据源和计划。数据从 InfluxDB 查询，由 query 节点计算，然后传递到数据管道中。

示例

batch
  |query('''
    SELECT mean("value")
    FROM "telegraf"."default".cpu_usage_idle
    WHERE "host" = 'serverA'
  ''')
    .period(1m)
    .every(20s)
    .groupBy(time(10s), 'cpu')
    ...

在上面的示例中，InfluxDB 每 20 秒查询一次；返回的时间窗口跨度为 1 分钟，并分组为 10 秒的桶。

要将 InfluxQL 高级语法用于 query 节点中的函数，您必须在 TICKScript 中表达部分 InfluxQL 查询。高级语法计算时间桶上的嵌套聚合，然后将外部聚合应用于结果。

例如，以下 InfluxQL 脚本计算每个 CPU 的 CPU 使用率 10 秒平均值的非负差值，对于 query 节点无效

SELECT non_negative_difference(mean("value"))
  FROM "telegraf"."default"."cpu_usage_idle"
  WHERE "host" = 'serverA'
  GROUP BY time(10s), "cpu"
  ...

要为 query 节点计算上述结果，您必须使用 TICKScript 指定分组和外部聚合

batch
  |query('''
    SELECT mean("value")
    FROM "telegraf"."default".cpu_usage_idle
    WHERE "host" = 'serverA'
  ''')
    .period(1m)
    .every(1m)
    .groupBy(time(10s), 'cpu')
  | difference('max_usage')
  | where(lambda: "difference" >= 0)
  ...

构造函数

链式方法描述
query ( q string)要执行的查询。WHERE 子句中不得包含时间条件，也不得包含 GROUP BY 子句。时间条件根据周期、偏移量和计划动态添加。GROUP BY 子句根据传递给 groupBy 方法的维度动态添加。

属性方法

设置器描述
align ( )使查询的开始和停止时间与 QueryNode 的偶数边界对齐。每个属性。如果使用 QueryNode.Cron 属性，则不适用。
alignGroup ( )使 group by 时间间隔与查询的开始时间对齐
cluster ( value string)配置的 InfluxDB 集群的名称。如果为空，将使用默认集群。
cron ( value string)使用 cron 语法定义计划。
every ( value time.Duration)查询 InfluxDB 的频率。
fill ( value interface{})填充数据。选项包括
groupBy ( d ...interface{})按一组维度对数据进行分组。可以指定一个时间维度。
groupByMeasurement ( )如果设置，将在组 ID 中包含度量名称。以及任何其他 group by 维度。
offset ( value time.Duration)从当前时间回溯查询的时间
period ( value time.Duration)将从 InfluxDB 查询的周期或时间长度
quiet ( )禁止来自此节点的所有错误日志记录事件。

链式方法

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

属性

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

Align

使查询的开始和停止时间与 QueryNode.Every 属性的偶数边界对齐。如果使用 QueryNode.Cron 属性，则不适用。

query.align()

AlignGroup

使 group by 时间间隔与查询的开始时间对齐

query.alignGroup()

Cluster

配置的 InfluxDB 集群的名称。如果为空，将使用默认集群。

query.cluster(value string)

Cron

使用 cron 语法定义计划。

具体的 cron 实现文档在此处：https://github.com/gorhill/cronexpr#implementation

Cron 属性与 Every 属性互斥。

query.cron(value string)

Every

查询 InfluxDB 的频率。

Every 属性与 Cron 属性互斥。

query.every(value time.Duration)

Fill

填充数据。选项包括

  • 任何数值
  • null - 表现出与默认值相同的行为
  • previous - 报告前一个窗口的值
  • none - 抑制值为 null 的时间戳和值
  • linear - 报告线性插值的结果
query.fill(value interface{})

GroupBy

按一组维度对数据进行分组。可以指定一个时间维度。

此属性向查询添加 GROUP BY 子句，因此当使用 GROUP BY 查询 InfluxDB 时，所有正常行为都适用。

当您的周期长于您的 group by 时间间隔时，请使用 group by 时间。

示例

    batch
        |query(...)
            .period(1m)
            .every(1m)
            .groupBy(time(10s), 'tag1', 'tag2'))
            .align()

group by 时间偏移也是可能的。

示例

    batch
        |query(...)
            .period(1m)
            .every(1m)
            .groupBy(time(10s, -5s), 'tag1', 'tag2'))
            .align()
            .offset(5s)

建议将 QueryNode.Align 和 QueryNode.Offset 与 group by 时间维度结合使用，以便时间边界与 group by 间隔匹配。要自动将 group by 间隔与查询时间的开始对齐，请使用 QueryNode.AlignGroup。 这在更复杂的情况下非常有用，例如当 groupBy 时间周期长于查询频率时。

示例

    batch
        |query(...)
            .period(5m)
            .every(30s)
            .groupBy(time(1m), 'tag1', 'tag2')
            .align()
            .alignGroup()

对于上面的示例，如果没有 QueryNode.AlignGroup， Kapacitor 发出的每个其他查询（在分钟后的 :30）将与 :00 秒对齐，而不是期望的 :30 秒，这将创建 6 个 group by 间隔而不是 5 个，其中第一个和最后一个将只有 30 秒的数据而不是完整的一分钟。如果在与 QueryNode.AlignGroup 结合使用 group by 时间偏移（即 time(t, offset)）时，将首先发生对齐，然后在指定的量之后偏移。

注意：由于 QueryNode.Offset 本质上是一个负属性，因此 “time” 函数的第二个 “offset” 参数为负数以匹配。

query.groupBy(d ...interface{})

GroupByMeasurement

如果设置，将在组 ID 中包含度量名称。以及任何其他 group by 维度。

示例

 batch
      |query('SELECT sum("value") FROM "telegraf"."autogen"./process_.*/')
          .groupByMeasurement()
          .groupBy('host')

上面的示例从匹配 `/process_.*/` 的多个度量中选择数据，然后每个点按主机标签和度量名称分组。因此，将度量保留在它们自己的组中。

query.groupByMeasurement()

Offset

从当前时间回溯查询的时间

例如，偏移量为 2 小时，Every 为 5 分钟，Kapacitor 将每 5 分钟查询 InfluxDB，获取 2 小时前的数据窗口。

这也适用于 Cron 计划。如果 cron 指定在每个星期日凌晨 1 点运行，并且偏移量为 1 小时。那么在星期日凌晨 1 点，将查询凌晨 12 点的数据。

query.offset(value time.Duration)

Period

将从 InfluxDB 查询的周期或时间长度

query.period(value time.Duration)

Quiet

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

query.quiet()

链式方法

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

Alert

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

query|alert()

返回：AlertNode

Barrier

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

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

query|barrier()

返回：BarrierNode

Bottom

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

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

返回：InfluxQLNode

ChangeDetect

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

query|changeDetect(field string)

返回：ChangeDetectNode

Combine

将此节点与其自身组合。数据在时间戳上组合。

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

返回：CombineNode

Count

计算点的数量。

query|count(field string)

返回：InfluxQLNode

CumulativeSum

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

query|cumulativeSum(field string)

返回：InfluxQLNode

Deadman

用于在低吞吐量（又名死人开关）上创建告警的辅助函数。

  • 阈值：如果吞吐量在点/间隔中降至阈值以下，则触发告警。
  • 间隔：检查吞吐量的频率。
  • 表达式：也要评估的可选表达式列表。对于一天中的时间告警很有用。

示例

    var data = batch
        |query()...
    // 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 = batch
        |query()...
    // 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 = batch
        |query()...
    // 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 = batch
        |query()...
    // 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...

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

返回：AlertNode

Default

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

query|default()

返回：DefaultNode

Delete

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

query|delete()

返回：DeleteNode

Derivative

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

query|derivative(field string)

返回：DerivativeNode

Difference

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

query|difference(field string)

返回：InfluxQLNode

Distinct

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

query|distinct(field string)

返回：InfluxQLNode

Ec2Autoscale

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

query|ec2Autoscale()

返回：Ec2AutoscaleNode

Elapsed

计算点之间经过的时间。

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

返回：InfluxQLNode

Eval

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

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

返回：EvalNode

First

选择第一个点。

query|first(field string)

返回：InfluxQLNode

Flatten

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

query|flatten()

返回：FlattenNode

HoltWinters

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

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

返回：InfluxQLNode

HoltWintersWithFit

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

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

query|httpOut(endpoint string)

返回：HTTPOutNode

HttpPost

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

query|httpPost(url ...string)

返回：HTTPPostNode

InfluxDBOut

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

query|influxDBOut()

返回：InfluxDBOutNode

Join

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

query|join(others ...Node)

返回：JoinNode

K8sAutoscale

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

query|k8sAutoscale()

返回：K8sAutoscaleNode

KapacitorLoopback

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

query|kapacitorLoopback()

返回：KapacitorLoopbackNode

Last

选择最后一个点。

query|last(field string)

返回：InfluxQLNode

Log

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

query|log()

返回：LogNode

Max

选择最大点。

query|max(field string)

返回：InfluxQLNode

Mean

计算数据的平均值。

query|mean(field string)

返回：InfluxQLNode

Median

计算数据的中位数。

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

query|median(field string)

返回：InfluxQLNode

Min

选择最小点。

query|min(field string)

返回：InfluxQLNode

Mode

计算数据的众数。

query|mode(field string)

返回：InfluxQLNode

MovingAverage

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

query|movingAverage(field string, window int64)

返回：InfluxQLNode

Percentile

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

query|percentile(field string, percentile float64)

返回：InfluxQLNode

Sample

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

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

query|sample(rate interface{})

返回：SampleNode

Shift

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

query|shift(shift time.Duration)

返回：ShiftNode

Sideload

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

query|sideload()

返回：SideloadNode

Spread

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

query|spread(field string)

返回：InfluxQLNode

StateCount

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

query|stateCount(expression ast.LambdaNode)

返回：StateCountNode

StateDuration

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

query|stateDuration(expression ast.LambdaNode)

返回：StateDurationNode

Stats

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

query|stats(interval time.Duration)

返回：StatsNode

Stddev

计算标准偏差。

query|stddev(field string)

返回：InfluxQLNode

Sum

计算所有值的总和。

query|sum(field string)

返回：InfluxQLNode

SwarmAutoscale

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

query|swarmAutoscale()

返回：SwarmAutoscaleNode

Top

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

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

返回：InfluxQLNode

Trickle

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

query|trickle()

返回：TrickleNode

Union

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

query|union(node ...Node)

返回：UnionNode

Where

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

query|where(expression ast.LambdaNode)

返回：WhereNode

Window

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

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

query|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 Core 和 Enterprise

快速启动。更快扩展。

获取更新

InfluxDB 3 Core 是一个开源、高速、最近数据引擎，可实时收集和处理数据，并将其持久化到本地磁盘或对象存储。InfluxDB 3 Enterprise 以 Core 的基础为构建，增加了高可用性、读取副本、增强的安全性以及数据压缩，从而实现更快的查询和优化的存储。InfluxDB 3 Enterprise 的免费层可供非商业家用或业余爱好者使用。

有关更多信息，请查看