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 秒的间隔分组。
要在 query 节点中使用 InfluxQL 高级语法处理函数，您必须在 TICKScript 中表达 InfluxQL 查询的一部分。高级语法在时间间隔上计算嵌套聚合，然后对结果应用外部聚合。
例如，以下 InfluxQL 脚本计算了每台 CPU 10 秒平均 CPU 使用率的非负差值，对于 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 子句。时间条件根据 period、offset 和 schedule 动态添加。GROUP BY 子句根据传递给 groupBy 方法的维度动态添加。 |
属性方法
| Setter 方法 | 描述 |
|---|---|
| align ( ) | 使 QueryNode.Every 属性的查询边界对齐查询的开始和停止时间。如果使用 QueryNode.Cron 属性，则不适用。 |
| alignGroup ( ) | 将组时间间隔与查询的开始时间对齐 |
cluster ( value string ) | 已配置的 InfluxDB 集群的名称。如果为空，则使用默认集群。 |
cron ( value string) | 使用 cron 语法定义计划。 |
every ( value time.Duration) | 查询 InfluxDB 的频率。 |
fill ( value interface{}) | 填充数据。选项有 |
groupBy ( d ...interface{}) | 按一组维度对数据进行分组。可以指定一个时间维度。 |
| groupByMeasurement ( ) | 如果设置为 true，则会将 measurement 名称包含在组 ID 中。以及任何其他分组的维度。 |
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
属性
属性方法会修改调用节点的状态。它们不会向管道添加新节点，并且始终返回对调用节点的引用。属性方法使用 . 运算符标记。
对齐
使 QueryNode.Every 属性的查询边界对齐查询的开始和停止时间。如果使用 QueryNode.Cron 属性，则不适用。
query.align()AlignGroup
将组时间间隔与查询的开始时间对齐
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 的正常行为都适用。
当您的周期长于分组时间间隔时，请使用按时间分组。
示例
batch
|query(...)
.period(1m)
.every(1m)
.groupBy(time(10s), 'tag1', 'tag2'))
.align()也可以进行时间偏移分组。
示例
batch
|query(...)
.period(1m)
.every(1m)
.groupBy(time(10s, -5s), 'tag1', 'tag2'))
.align()
.offset(5s)建议将 QueryNode.Align 和 QueryNode.Offset 与时间分组维度结合使用，以便时间边界与分组间隔匹配。要自动将分组间隔与查询时间开始时间对齐，请使用 QueryNode.AlignGroup。 这在更复杂的情况下很有用，例如当分组时间周期长于查询频率时。
示例
batch
|query(...)
.period(5m)
.every(30s)
.groupBy(time(1m), 'tag1', 'tag2')
.align()
.alignGroup()对于上面的示例，如果没有 QueryNode.AlignGroup， Kapacitor 发出的每隔一次查询（在分钟的 :30 分）将与 :00 秒对齐，而不是预期的 :30 秒，这将创建 6 个分组间隔而不是 5 个，其中第一个和最后一个将只有 30 秒的数据，而不是一整分钟。如果时间偏移分组（即 time(t, offset)）与 QueryNode.AlignGroup 结合使用， 则首先进行对齐，然后根据指定的数量进行偏移。
注意：由于 QueryNode.Offset 本质上是一个负属性，因此“time”函数中的第二个“offset”参数为负值以匹配。
query.groupBy(d ...interface{})GroupByMeasurement
如果设置为 true，则会将 measurement 名称包含在组 ID 中。以及任何其他分组的维度。
示例
batch
|query('SELECT sum("value") FROM "telegraf"."autogen"./process_.*/')
.groupByMeasurement()
.groupBy('host')上面的示例从匹配 `/process_.*` 的几个度量中选择数据，然后每个点都按主机标签和度量名称进行分组。因此，保持度量在各自的组中。
query.groupByMeasurement()Offset
从当前时间开始向前查询数据的时间范围
例如，Offset 为 2 小时，Every 为 5 分钟，Kapacitor 将每 5 分钟查询一次 InfluxDB，以获取 2 小前的数据窗口。
这也适用于 Cron 计划。如果 cron 指定每周日凌晨 1 点运行，Offset 为 1 小时。那么在周日凌晨 1 点，将查询 12 点的数据。
query.offset(value time.Duration)周期
从 InfluxDB 查询的期间或时间长度
query.period(value time.Duration)Quiet
抑制此节点的所有错误日志事件。
query.quiet()链式方法
链式方法在管道中创建一个新的节点作为调用节点的子节点。它们不会修改调用节点。链式方法使用 | 运算符标记。
警报（Alert）
创建一个警报节点，可以触发警报。
query|alert()返回：AlertNode
Barrier
创建一个新的 Barrier 节点，该节点会定期发出 BarrierMessage。
每个 period duration 将发出一个 BarrierMessage。
query|barrier()返回：BarrierNode
Bottom
选择 field 的底部 num 个点，并按任何额外的标签或字段排序。
query|bottom(num int64, field string, fieldsAndTags ...string)返回：InfluxQLNode
ChangeDetect
创建一个新节点，该节点仅在点与前一个点不同时才发出新点。
query|changeDetect(field string)Combine
将此节点与自身合并。数据按时间戳合并。
query|combine(expressions ...ast.LambdaNode)返回：CombineNode
Count
计算点的数量。
query|count(field string)返回：InfluxQLNode
CumulativeSum
计算接收到的每个点的累积总和。对于收集到的每个点都会发出一个点。
query|cumulativeSum(field string)返回：InfluxQLNode
死人开关（Deadman）
用于创建低吞吐量警报（即死人开关）的辅助函数。
- 阈值：如果吞吐量低于每 points/interval 的点数，则触发警报。
- 间隔：检查吞吐量的频率。
- 表达式：可选的表达式列表，也用于评估。对于按时警报很有用。
示例
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
默认
创建一个可以为缺失的标签或字段设置默认值的节点。
query|default()返回：DefaultNode
删除
创建一个可以删除标签或字段的节点。
query|delete()返回：DeleteNode
导数
创建一个新节点，该节点计算相邻点的导数。
query|derivative(field string)Difference
计算点之间的差异，与经过的时间无关。
query|difference(field string)返回：InfluxQLNode
Distinct
生成仅包含不同点的批次。
query|distinct(field string)返回：InfluxQLNode
Ec2Autoscale
创建一个可以为 ec2 自动伸缩组触发自动伸缩事件的节点。
query|ec2Autoscale()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 个参数，则必须指定一个 endpoint 属性方法。
query|httpPost(url ...string)返回：HTTPPostNode
InfluxDBOut
创建一个 influxdb 输出节点，该节点会将传入的数据存储到 InfluxDB 中。
query|influxDBOut()Join
将此节点与其他节点连接。数据按时间戳连接。
query|join(others ...Node)返回：JoinNode
K8sAutoscale
创建一个可以为 kubernetes 集群触发自动伸缩事件的节点。
query|k8sAutoscale()KapacitorLoopback
创建一个 kapacitor 回送节点，该节点会将数据作为流重新发送到 Kapacitor 中。
query|kapacitorLoopback()Last
选择最后一个点。
query|last(field string)返回：InfluxQLNode
Log
创建一个记录其接收的所有数据的节点。
query|log()返回：LogNode
Max
选择最大点。
query|max(field string)返回：InfluxQLNode
平均值
计算数据的平均值。
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
计算最后 window 个点的移动平均值。直到窗口填满才会发出点。
query|movingAverage(field string, window int64)返回：InfluxQLNode
Percentile
选择给定百分位数的点。这是一个选择器函数，不会在点之间进行插值。
query|percentile(field string, percentile float64)返回：InfluxQLNode
Sample
创建一个新节点，该节点对传入的点或批次进行采样。
将发出一个点，每隔指定的 count 或 duration。
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)StateDuration
创建一个节点，该节点跟踪给定状态下的持续时间。
query|stateDuration(expression ast.LambdaNode)Stats
创建一个新的数据流，其中包含节点的内部统计信息。interval 表示基于实际时间发出统计信息的频率。这意味着 interval 时间与源节点接收的数据点的时间无关。
query|stats(interval time.Duration)返回：StatsNode
Stddev
计算标准偏差。
query|stddev(field string)返回：InfluxQLNode
Sum
计算所有值的总和。
query|sum(field string)返回：InfluxQLNode
SwarmAutoscale
创建一个可以为 Docker swarm 集群触发自动伸缩事件的节点。
query|swarmAutoscale()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 只能应用于流边缘。
query|window()返回：WindowNode
此页面是否有帮助？
感谢您的反馈！
支持和反馈
感谢您成为我们社区的一员！我们欢迎并鼓励您对 Kapacitor 和本文档提供反馈和错误报告。要获取支持，请使用以下资源：