BarrierNode
barrier
节点基于以下其中一项发出 barrier
- 自上次接收到数据点以来的空闲时间
- 基于系统时间的周期性计时器
Barrier 使您可以在没有数据流量的情况下执行管道。在指定的 barrier 之后接收到的数据点将被丢弃。
基于空闲时间的 Barrier 示例
stream
|from()
.measurement('cpu')
|barrier()
.idle(11s) // Must be longer than the window period
.delete(TRUE)
|window()
.period(10s)
.every(5s)
|top(10, 'value')
// Post the top 10 results over the last 10s updated every 5s.
|httpPost('http://example.com/api/top10')
在 .delete(TRUE)
中,TRUE
必须为大写。
构造函数
链式方法 | 描述 |
---|---|
barrier( ) | 创建一个新的 Barrier 节点,该节点定期发出 BarrierMessage |
属性方法
设置器 | 描述 |
---|---|
idle ( value time.Duration ) | 基于自上次接收到消息以来的空闲时间发出 barrier。必须大于零且长于窗口 period 。 |
period ( value time.Duration ) | 基于周期性计时器发出 barrier。计时器基于系统时钟而不是消息时间。必须大于零。 |
delete ( value Boolean ) | 在处理每个 barrier 后删除组。 |
quiet ( ) | 禁止来自此节点的所有错误日志记录事件。 |
链式方法
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
属性
属性方法修改调用节点上的状态。它们不会向管道添加另一个节点,并且始终返回对调用节点的引用。属性方法使用 .
运算符标记。
Idle
基于自上次接收到消息以来的空闲时间发出 barrier。必须大于零且长于窗口 period
。
barrier.idle(value time.Duration)
Barrier 空闲时间和窗口周期
idle
必须大于关联的 window 的 period
。如果 idle
时间小于窗口 period
,则可能会丢失数据。
Period
基于周期性计时器发出 barrier。计时器基于系统时钟而不是消息时间。必须大于零。
barrier.period(value time.Duration)
Delete
Delete 指示应在处理每个 barrier 后删除组。这包括 barrier 节点本身,这意味着如果 delete
为 true
,则每个组仅触发一次 barrier,并且 barrier 节点忘记该组。如果为该组收到新点,则将再次创建该组。
如果您随时间推移增加基数,这将非常有用,因为一旦为组触发 barrier,它将被删除,从而释放任何管理该组的资源。
barrier.delete()
delete
将释放用于管理组的系统资源,并有助于保持 Kapacitor 的整体性能,但这些收益是最小的。有关优化任务的信息,请参阅 如何优化 Kapacitor 任务?
Quiet
禁止来自此节点的所有错误日志记录事件。
barrier.quiet()
链式方法
链式方法在管道中创建一个新节点作为调用节点的子节点。它们不会修改调用节点。链式方法使用 |
运算符标记。
Alert
创建一个警报节点,该节点可以触发警报。
barrier|alert()
返回值:AlertNode
Barrier
创建一个新的 Barrier 节点,该节点定期发出 BarrierMessage。
每个 period 持续时间将发出一个 BarrierMessage。
barrier|barrier()
返回值:BarrierNode
Bottom
选择 field
的底部 num
个点,并按任何额外的标签或字段排序。
barrier|bottom(num int64, field string, fieldsAndTags ...string)
返回值:InfluxQLNode
ChangeDetect
创建一个新节点,该节点仅在与上一个点不同时才发出新点。
barrier|changeDetect(field string)
返回值:ChangeDetectNode
Combine
将此节点与其自身组合。数据按时间戳组合。
barrier|combine(expressions ...ast.LambdaNode)
返回值:CombineNode
Count
计算点的数量。
barrier|count(field string)
返回值:InfluxQLNode
CumulativeSum
计算接收到的每个点的累积总和。为收集的每个点发出一个点。
barrier|cumulativeSum(field string)
返回值:InfluxQLNode
Deadman
用于在低吞吐量(又名 deadman’s switch)上创建警报的辅助函数。
- 阈值:如果吞吐量在点/间隔中降至阈值以下,则触发警报。
- 间隔:检查吞吐量的频率。
- 表达式:也要评估的可选表达式列表。对一天中的时间警报很有用。
示例
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...
barrier|deadman(threshold float64, interval time.Duration, expr ...ast.LambdaNode)
返回值:AlertNode
Default
创建一个节点,该节点可以为缺少的标签或字段设置默认值。
barrier|default()
返回值:DefaultNode
Delete
创建一个节点,该节点可以删除标签或字段。
barrier|delete()
返回值:DeleteNode
Derivative
创建一个新节点,该节点计算相邻点的导数。
barrier|derivative(field string)
返回值:DerivativeNode
Difference
计算点之间的差异,与经过的时间无关。
barrier|difference(field string)
返回值:InfluxQLNode
Distinct
生成仅包含不同点的批次。
barrier|distinct(field string)
返回值:InfluxQLNode
Ec2Autoscale
创建一个节点,该节点可以为 ec2 autoscalegroup 触发自动缩放事件。
barrier|ec2Autoscale()
返回值:Ec2AutoscaleNode
Elapsed
计算点之间经过的时间。
barrier|elapsed(field string, unit time.Duration)
返回值:InfluxQLNode
Eval
创建一个 eval 节点,该节点将评估给定变换函数到每个数据点。可以提供表达式列表,并将按给定的顺序评估表达式。结果可用于以后的表达式。
barrier|eval(expressions ...ast.LambdaNode)
返回值:EvalNode
First
选择第一个点。
barrier|first(field string)
返回值:InfluxQLNode
Flatten
将具有相似时间的点展平为单个点。
barrier|flatten()
返回值:FlattenNode
GroupBy
按一组标签对数据进行分组。
可以传递文字 * 以按所有维度分组。示例
|groupBy(*)
barrier|groupBy(tag ...interface{})
返回值:GroupByNode
HoltWinters
计算数据集的 Holt-Winters (/influxdb/v1/query_language/functions/#holt-winters) 预测。
barrier|holtWinters(field string, h int64, m int64, interval time.Duration)
返回值:InfluxQLNode
HoltWintersWithFit
计算数据集的 Holt-Winters (/influxdb/v1/query_language/functions/#holt-winters) 预测。此方法还输出用于拟合数据的所有点以及预测数据。
barrier|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
请求数据。
barrier|httpOut(endpoint string)
返回值:HTTPOutNode
HttpPost
创建一个 HTTP Post 节点,该节点将接收到的数据 POST 到提供的 HTTP 端点。HttpPost 期望 0 或 1 个参数。如果提供 0 个参数,则必须指定端点属性方法。
barrier|httpPost(url ...string)
返回值:HTTPPostNode
InfluxDBOut
创建一个 influxdb 输出节点,该节点将传入数据存储到 InfluxDB 中。
barrier|influxDBOut()
返回值:InfluxDBOutNode
Join
将此节点与其他节点连接。数据按时间戳连接。
barrier|join(others ...Node)
返回值:JoinNode
K8sAutoscale
创建一个节点,该节点可以为 kubernetes 集群触发自动缩放事件。
barrier|k8sAutoscale()
返回值:K8sAutoscaleNode
KapacitorLoopback
创建一个 kapacitor loopback 节点,该节点将数据作为流发送回 Kapacitor。
barrier|kapacitorLoopback()
Last
选择最后一个点。
barrier|last(field string)
返回值:InfluxQLNode
Log
创建一个节点,该节点记录它接收的所有数据。
barrier|log()
返回值:LogNode
Max
选择最大点。
barrier|max(field string)
返回值:InfluxQLNode
Mean
计算数据的平均值。
barrier|mean(field string)
返回值:InfluxQLNode
Median
计算数据的中位数。
注意: 此方法不是选择器。如果您想要中位数点,请使用
.percentile(field, 50.0)
。
barrier|median(field string)
返回值:InfluxQLNode
Min
选择最小点。
barrier|min(field string)
返回值:InfluxQLNode
Mode
计算数据的众数。
barrier|mode(field string)
返回值:InfluxQLNode
MovingAverage
计算最后窗口点的移动平均值。在窗口填满之前,不会发出任何点。
barrier|movingAverage(field string, window int64)
返回值:InfluxQLNode
Percentile
选择给定百分位数的点。这是一个选择器函数,不执行点之间的插值。
barrier|percentile(field string, percentile float64)
返回值:InfluxQLNode
Sample
创建一个新节点,该节点对传入的点或批次进行采样。
将每隔指定的计数或持续时间发出一个点。
barrier|sample(rate interface{})
返回值:SampleNode
Shift
创建一个新节点,该节点及时移动传入的点或批次。
barrier|shift(shift time.Duration)
返回值:ShiftNode
Sideload
创建一个可以从外部源加载数据的节点。
barrier|sideload()
返回值:SideloadNode
Spread
计算 min
和 max
点之间的差值。
barrier|spread(field string)
返回值:InfluxQLNode
StateCount
创建一个节点,该节点跟踪给定状态下连续点的数量。
barrier|stateCount(expression ast.LambdaNode)
返回值:StateCountNode
StateDuration
创建一个节点,该节点跟踪给定状态的持续时间。
barrier|stateDuration(expression ast.LambdaNode)
Stats
创建一个新的数据流,其中包含节点的内部统计信息。间隔表示基于实时发出统计信息的频率。这意味着间隔时间独立于源节点正在接收的数据点的时间。
barrier|stats(interval time.Duration)
返回值:StatsNode
Stddev
计算标准差。
barrier|stddev(field string)
返回值:InfluxQLNode
Sum
计算所有值的总和。
barrier|sum(field string)
返回值:InfluxQLNode
SwarmAutoscale
创建一个节点,该节点可以为 Docker swarm 集群触发自动缩放事件。
barrier|swarmAutoscale()
Top
选择 field
的顶部 num
个点,并按任何额外的标签或字段排序。
barrier|top(num int64, field string, fieldsAndTags ...string)
返回值:InfluxQLNode
Trickle
创建一个新节点,该节点将批处理数据转换为流数据。
barrier|trickle()
返回值:TrickleNode
Union
执行此节点和所有其他给定节点的联合。
barrier|union(node ...Node)
返回值:UnionNode
Where
创建一个新节点,该节点通过给定的表达式过滤数据流。
barrier|where(expression ast.LambdaNode)
返回值:WhereNode
Window
创建一个新节点,该节点按时间窗口化流。
注意:Window 只能应用于流边缘。
barrier|window()
返回值:WindowNode
此页是否对您有帮助?
感谢您的反馈!
支持和反馈
感谢您成为我们社区的一份子!我们欢迎并鼓励您提供关于 Kapacitor 和本文档的反馈和错误报告。要获得支持,请使用以下资源