文档

WindowNode

window 节点在移动时间范围内缓存数据。windowperiod 属性定义了 window 覆盖的时间范围。

windowevery 属性定义了窗口向管道中下一个节点发出的频率。

windowalign 属性定义了如何对齐窗口边缘。(默认情况下,边缘是相对于 window 节点接收的第一个数据点定义的。)

示例

stream
  |window()
    .period(10m)
    .every(5m)
  |httpOut('recent')

此示例每 5 分钟 将最近 10 分钟 的周期发送到管道的 httpOut 节点。由于 every 小于 period,因此每次发出窗口时,它都包含 5 分钟 的新数据和 5 分钟 的上一个周期的数据。

注意: 由于未定义 align 属性,因此 window 边缘是相对于第一个数据点定义的。

构造函数

链式方法描述
window ( )创建一个按时间窗口化数据流的新节点。

属性方法

设置器描述
align ( )如果未使用 align 属性来修改 window 节点,则假定窗口对齐从其接收的第一个数据点的时间开始。如果设置了 align 属性,则窗口时间边缘将被截断为 every 属性(例如,如果数据点的时间为 12:06 且 every 属性为 5m,则数据点的窗口范围将从 12:05 到 12:10)。
every ( value time.Duration)当前窗口向管道发出的频率。如果等于零,则每个新点都将发出当前窗口。
everyCount ( value int64)EveryCount 确定基于点计数发出窗口的频率。值为 1 表示每个新点都将发出窗口。
fillPeriod ( )FillPeriod 指示 WindowNode 等待 period 经过后再发出第一个批次。这仅在 period 大于 every 值时适用。
period ( value time.Duration)窗口的周期或时间长度。
periodCount ( value int64)PeriodCount 是每个窗口的点数。
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


属性

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

对齐

设置 align 属性以将窗口时间边缘截断为 every 属性。例如,如果数据点的时间为 12:06 且 every 属性为 5m,则数据点的窗口范围为 12:05 到 12:10)。

如果 align 属性未用于修改 window 节点,则窗口对齐从接收到第一个数据点的时间开始。

window.align()

注意:当以不规则的时间间隔摄取数据时,我们建议使用 window.align() 来对数据进行分组。

Every

当前窗口向管道发出的频率。如果等于零,则每个新点都将发出当前窗口。

window.every(value time.Duration)

EveryCount

EveryCount 确定基于点计数发出窗口的频率。值为 1 表示每个新点都将发出窗口。

window.everyCount(value int64)

FillPeriod

FillPeriod 指示 WindowNode 等待 period 经过后再发出第一个批次。这仅在 period 大于 every 值时适用。

window.fillPeriod()

Period

窗口的周期或时间长度。

window.period(value time.Duration)

PeriodCount

PeriodCount 是每个窗口的点数。

window.periodCount(value int64)

Quiet

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

window.quiet()

链式方法

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

Alert

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

window|alert()

返回:AlertNode

Barrier

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

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

window|barrier()

返回:BarrierNode

Bottom

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

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

返回:InfluxQLNode

ChangeDetect

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

window|changeDetect(field string)

返回:ChangeDetectNode

Combine

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

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

返回:CombineNode

Count

计算点数。

window|count(field string)

返回:InfluxQLNode

CumulativeSum

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

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

idmessage 告警属性可以通过 ‘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...
window|deadman(threshold float64, interval time.Duration, expr ...ast.LambdaNode)

返回:AlertNode

Default

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

window|default()

返回:DefaultNode

Delete

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

window|delete()

返回:DeleteNode

Derivative

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

window|derivative(field string)

返回:DerivativeNode

Difference

计算点之间的差值,与经过的时间无关。

window|difference(field string)

返回:InfluxQLNode

Distinct

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

window|distinct(field string)

返回:InfluxQLNode

Ec2Autoscale

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

window|ec2Autoscale()

返回:Ec2AutoscaleNode

Elapsed

计算点之间经过的时间。

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

返回:InfluxQLNode

Eval

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

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

返回:EvalNode

First

选择第一个点。

window|first(field string)

返回:InfluxQLNode

Flatten

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

window|flatten()

返回:FlattenNode

GroupBy

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

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

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

返回:GroupByNode

HoltWinters

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

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

返回:InfluxQLNode

HoltWintersWithFit

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

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

window|httpOut(endpoint string)

返回:HTTPOutNode

HttpPost

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

window|httpPost(url ...string)

返回:HTTPPostNode

InfluxDBOut

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

window|influxDBOut()

返回:InfluxDBOutNode

Join

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

window|join(others ...Node)

返回:JoinNode

K8sAutoscale

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

window|k8sAutoscale()

返回:K8sAutoscaleNode

KapacitorLoopback

创建一个 kapacitor loopback 节点,该节点将数据作为流发送回 Kapacitor。

window|kapacitorLoopback()

返回:KapacitorLoopbackNode

Last

选择最后一个点。

window|last(field string)

返回:InfluxQLNode

日志

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

window|log()

返回:LogNode

Max

选择最大点。

window|max(field string)

返回:InfluxQLNode

Mean

计算数据的平均值。

window|mean(field string)

返回:InfluxQLNode

Median

计算数据的中位数。

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

window|median(field string)

返回:InfluxQLNode

Min

选择最小点。

window|min(field string)

返回:InfluxQLNode

Mode

计算数据的众数。

window|mode(field string)

返回:InfluxQLNode

MovingAverage

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

window|movingAverage(field string, window int64)

返回:InfluxQLNode

Percentile

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

window|percentile(field string, percentile float64)

返回:InfluxQLNode

Sample

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

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

window|sample(rate interface{})

返回:SampleNode

Shift

创建一个新节点,该节点及时移动传入的点或批次。

window|shift(shift time.Duration)

返回:ShiftNode

Sideload

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

window|sideload()

返回:SideloadNode

Spread

计算 minmax 点之间的差值。

window|spread(field string)

返回:InfluxQLNode

StateCount

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

window|stateCount(expression ast.LambdaNode)

返回:StateCountNode

StateDuration

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

window|stateDuration(expression ast.LambdaNode)

返回:StateDurationNode

Stats

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

window|stats(interval time.Duration)

返回:StatsNode

Stddev

计算标准差。

window|stddev(field string)

返回:InfluxQLNode

Sum

计算所有值的总和。

window|sum(field string)

返回:InfluxQLNode

SwarmAutoscale

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

window|swarmAutoscale()

返回:SwarmAutoscaleNode

Top

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

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

返回:InfluxQLNode

Trickle

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

window|trickle()

返回:TrickleNode

Union

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

window|union(node ...Node)

返回:UnionNode

Where

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

window|where(expression ast.LambdaNode)

返回:WhereNode

Window

创建一个按时间窗口化数据流的新节点。

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

window|window()

返回:WindowNode


此页面是否对您有帮助?

感谢您的反馈!


Flux 的未来

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

阅读更多

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

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

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

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

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