文档文档

StreamNode

stream 节点表示通过其任何输入流式传输到 Kapacitor 的数据源。流任务中的 stream 变量是 StreamNode 的实例。StreamNode.From 是此节点的方法/属性。

构造函数

链式方法描述
stream没有构造函数签名。

属性方法

设置器描述
quiet ( )禁止来自此节点的所有错误日志记录事件。

链式方法

Deadman, From, Stats


属性

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

Quiet

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

stream.quiet()

链式方法

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

Deadman

用于在低吞吐量(又名“死人开关”)时创建警报的辅助函数。

  • 阈值:如果吞吐量在 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...

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 表达式,以进一步约束何时触发死人开关。示例

    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...
stream|deadman(threshold float64, interval time.Duration, expr ...ast.LambdaNode)

返回:AlertNode

From

创建一个新的 FromNode,可以使用 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')
        ...
stream|from()

返回:FromNode

Stats

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

stream|stats(interval time.Duration)

返回:StatsNode


此页内容对您有帮助吗?

感谢您的反馈!


Flux 的未来

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

阅读更多

现已全面上市

InfluxDB 3 Core 和 Enterprise

快速启动。更快扩展。

获取更新

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

有关更多信息,请查看