文档文档

JoinNode

join 节点连接来自任意数量节点的数据。当从父节点接收到每个数据点时,它会与来自其他父节点的具有匹配时间戳的下一个数据点配对。每个父节点最多为每个连接点贡献一个点。可以提供容差来连接时间戳不完全对齐的点。任何落在容差范围内的点都将按时间戳连接。如果多个点落在同一容差窗口内,则它们将按照到达顺序连接。

别名用于为来自各个节点的所有字段添加前缀。

连接可以是内部连接或外部连接,请参阅 JoinNode.Fill 属性。

示例:连接两个指标

在下面的示例中,errorsrequests 流被连接并转换以计算组合字段。

var errors = stream
  |from()
    .measurement('errors')
var requests = stream
  |from()
    .measurement('requests')
// Join the errors and requests streams
errors
  |join(requests)
    // Provide prefix names for the fields of the data points.
    .as('errors', 'requests')
    // points that are within 1 second are considered the same time.
    .tolerance(1s)
    // fill missing values with 0, implies outer join.
    .fill(0.0)
    // name the resulting stream
    .streamName('error_rate')
    // treat a delete from one side of the join as a delete to all sides
    .deleteAll(TRUE)
  // Both the "value" fields from each parent have been prefixed
  // with the respective names 'errors' and 'requests'.
  |eval(lambda: "errors.value" / "requests.value")
    .as('rate')
  ...

示例:连接三个或更多指标

在下面的示例中,errorsmissing_page_errorsserver_errors 被连接并转换以计算两个组合字段:404_rate500_rate

var errors = stream
|from()
  .measurement('errors')

var missing_page_errors = stream
|from()
  .measurement('errors')
  .where(lambda: "type" == '404')

var server_errors = stream
|from()
  .measurement('errors')
  .where(lambda: "type" == '500')

// Join the errors, missing_page_errors, and server_errors streams
errors
  |join(missing_page_errors, server_errors)
    // Provide prefix names for the fields of the data points.
    .as('errors', '404', '500')
    // points that are within 1 second are considered the same time.
    .tolerance(1s)
    // fill missing values with 0, implies outer join.
    .fill(0.0)
    // name the resulting stream
    .streamName('error_rates')
  // The "value" fields from each parent have been prefixed
  // with the respective names 'errors', 'missing_page_errors', 'and server_errors'.
  // Calculate the percentage of 404 errors
  |eval(lambda: "404.value" / "errors.value")
    .as('404_rate')
    // Calculate the percentage of 500 errors
  |eval(lambda: "500.value" / "errors.value")
    .as('500_rate')
  ...

构造函数

链式方法描述
join(others ...Node)将此节点与其他节点连接。数据按时间戳连接。

属性方法

设置器描述
as(names ...string)为来自各个节点的所有字段添加前缀名称。来自父节点的每个字段都将以提供的名称和一个 . 为前缀。请参阅下面的示例。
delimiter(value string)字段名称前缀的分隔符。可以是空字符串。
deleteAll(value bool)删除连接的两侧,无论哪一侧接收到删除消息。
fill(value interface{})填充数据。fill 选项暗示了连接的类型:内部连接或完全外部连接。
on(dims ...string)在分组依据维度的子集上连接。这是一种特殊情况,您希望来自一个父节点的单个点与来自不同父节点的多个点连接。
quiet()禁止来自此节点的所有错误日志记录事件。
streamName(value string)此新连接数据流的名称。如果为空,则使用左侧父节点的名称。
tolerance(value time.Duration)两个传入点之间可以相隔的最长时间,并且仍然被认为在时间上相等。连接的数据点的时间将四舍五入到最接近的容差持续时间的倍数。

链式方法

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


属性

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

As

为来自各个节点的所有字段添加前缀名称。来自父节点的每个字段都将以提供的名称和一个 ‘.’ 为前缀。请参阅上面的示例。

名称不能包含点 ‘.’ 字符。

join.as(names ...string)

Delimiter

字段名称前缀的分隔符。可以是空字符串。

join.delimiter(value string)

DeleteAll

删除连接的两侧,无论哪一侧接收到删除消息。

join.deleteAll(value bool)

Fill

填充数据。fill 选项暗示了连接的类型:内部连接或完全外部连接。选项有

  • none - (默认)跳过缺少点的行,内部连接。
  • null - 用 null 填充缺失点,完全外部连接。
  • 任何数值 - 用给定值填充字段,完全外部连接。

当使用数值或 null 填充时,字段名称通过从另一个点复制字段名称来确定。当不同的源具有不同的字段名称时,这效果不佳。如果需要,请使用 DefaultNodeDeleteNode 来完成填充操作。

join.fill(value interface{})

示例

    var maintlock = stream
        |from()
            .measurement('maintlock')
            .groupBy('service')
    var requests = stream
        |from()
            .measurement('requests')
            .groupBy('service')
    // Join the maintlock and requests streams
    // The intent it to drop any points in maintenance mode.
    maintlock
        |join(requests)
            // Provide prefix names for the fields of the data points.
            .as('maintlock', 'requests')
            // points that are within 1 second are considered the same time.
            .tolerance(1s)
            // fill missing fields with null, implies outer join.
            // a better default per field will be set later.
            .fill('null')
            // name the resulting stream.
            .streamName('requests')
        |default()
            // default maintenance mode to false, overwriting the null value if present.
            .field('maintlock.mode', false)
            // default the requests to 0, again overwriting the null value if present.
            .field('requests.value', 0.0)
        // drop any points that are in maintenance mode.
        |where(lambda: "maintlock.mode")
        |...

处理外部连接中的 null 填充值

当使用 Kapacitor 执行外部连接时,为连接和填充操作产生的 null 字段设置默认值非常重要。这是使用 DefaultNode 完成的,它将特定字段键的 null 值替换为指定的默认值。如果不这样做,可能会导致无效的行协议(因为 null 不是所有字段类型的合适值),从而导致连接失败。

source1
  |join(source2)
    .as('source1', 'source2')
    .fill('null')
  |default()
    // .field('field-key', default-value)

    // Define a default for an integer field type
    .field('source1.rounded', 0)
    // Define a default for a float field type
    .field('source1.value', 0.0)
    // Define a default for a string field type
    .field('source2.location', '')
    // Define a default for a boolean field type
    .field('source2.maintenance', false)

当使用此方法时,您必须了解连接产生的所有字段和字段类型,并提供适当的默认值。

您还可以使用 DeleteNode 删除连接产生的不必要的字段或标签。

source1
  |join(source2)
    .as('source1', 'source2')
    .fill('null')
  |default()
    .field('source1.mode', false)
    .field('source2.value', 0.0)
  |delete()
    .field('source1.anon')
    .tag('host')

On

在分组依据维度的子集上连接。这是一种特殊情况,您希望来自一个父节点的单个点与来自不同父节点的多个点连接。

例如,给定两个指标

  1. building_power(单个值):按建筑标记,值是建筑物消耗的总功率。
  2. floor_power(多个值):按建筑和楼层标记,值是每层楼消耗的总功率。

您想计算每层楼消耗的总建筑功率的百分比。由于每个建筑物只有一个点,因此您需要将其与来自每层楼的点多次连接。通过将 on 维度定义为 building,我们表示我们希望仅具有 building 标签的点与具有更多标签的更具体的点连接,在本例中为 floor 标签。换句话说,当我们有带有标签 building 和 floor 的点时,我们只想在 building 标签上连接。

示例

    var building = stream
        |from()
            .measurement('building_power')
            .groupBy('building')
    var floor = stream
        |from()
            .measurement('floor_power')
            .groupBy('building', 'floor')
    building
        |join(floor)
            .as('building', 'floor')
            .on('building')
        |eval(lambda: "floor.value" / "building.value")
            ... // Values here are grouped by 'building' and 'floor'
join.on(dims ...string)

Quiet

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

join.quiet()

StreamName

此新连接数据流的名称。如果为空,则使用左侧父节点的名称。

join.streamName(value string)

Tolerance

两个传入点之间可以相隔的最长时间,并且仍然被认为在时间上相等。连接的数据点的时间将四舍五入到最接近的容差持续时间的倍数。

join.tolerance(value time.Duration)

链式方法

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

Alert

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

join|alert()

返回:AlertNode

Barrier

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

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

join|barrier()

返回:BarrierNode

Bottom

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

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

返回:InfluxQLNode

ChangeDetect

创建一个新节点,该节点仅在与上一个点不同的情况下才发出新点。

join|changeDetect(field string)

返回:ChangeDetectNode

Combine

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

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

返回:CombineNode

Count

计算点的数量。

join|count(field string)

返回:InfluxQLNode

CumulativeSum

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

join|cumulativeSum(field string)

返回:InfluxQLNode

Deadman

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

  • Threshold:如果吞吐量降至阈值以下(点/间隔),则触发警报。
  • Interval:检查吞吐量的频率。
  • Expressions:可选的表达式列表,也用于评估。对于一天中的时间警报很有用。

示例

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

返回:AlertNode

Default

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

join|default()

返回:DefaultNode

Delete

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

join|delete()

返回:DeleteNode

Derivative

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

join|derivative(field string)

返回:DerivativeNode

Difference

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

join|difference(field string)

返回:InfluxQLNode

Distinct

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

join|distinct(field string)

返回:InfluxQLNode

Ec2Autoscale

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

join|ec2Autoscale()

返回:Ec2AutoscaleNode

Elapsed

计算点之间经过的时间。

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

返回:InfluxQLNode

Eval

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

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

返回:EvalNode

First

选择第一个点。

join|first(field string)

返回:InfluxQLNode

Flatten

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

join|flatten()

返回:FlattenNode

GroupBy

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

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

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

返回:GroupByNode

HoltWinters

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

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

返回:InfluxQLNode

HoltWintersWithFit

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

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

join|httpOut(endpoint string)

返回:HTTPOutNode

HttpPost

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

join|httpPost(url ...string)

返回:HTTPPostNode

InfluxDBOut

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

join|influxDBOut()

返回:InfluxDBOutNode

Join

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

join|join(others ...Node)

返回:JoinNode

K8sAutoscale

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

join|k8sAutoscale()

返回:K8sAutoscaleNode

KapacitorLoopback

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

join|kapacitorLoopback()

返回:KapacitorLoopbackNode

Last

选择最后一个点。

join|last(field string)

返回:InfluxQLNode

Log

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

join|log()

返回:LogNode

Max

选择最大点。

join|max(field string)

返回:InfluxQLNode

Mean

计算数据的平均值。

join|mean(field string)

返回:InfluxQLNode

Median

计算数据的中位数。

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

join|median(field string)

返回:InfluxQLNode

Min

选择最小点。

join|min(field string)

返回:InfluxQLNode

Mode

计算数据的众数。

join|mode(field string)

返回:InfluxQLNode

MovingAverage

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

join|movingAverage(field string, window int64)

返回:InfluxQLNode

Percentile

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

join|percentile(field string, percentile float64)

返回:InfluxQLNode

Sample

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

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

join|sample(rate interface{})

返回:SampleNode

Shift

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

join|shift(shift time.Duration)

返回:ShiftNode

Sideload

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

join|sideload()

返回:SideloadNode

Spread

计算 minmax 点之间的差值。

join|spread(field string)

返回:InfluxQLNode

StateCount

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

join|stateCount(expression ast.LambdaNode)

返回:StateCountNode

StateDuration

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

join|stateDuration(expression ast.LambdaNode)

返回:StateDurationNode

Stats

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

join|stats(interval time.Duration)

返回:StatsNode

Stddev

计算标准偏差。

join|stddev(field string)

返回:InfluxQLNode

Sum

计算所有值的总和。

join|sum(field string)

返回:InfluxQLNode

SwarmAutoscale

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

join|swarmAutoscale()

返回:SwarmAutoscaleNode

Top

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

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

返回:InfluxQLNode

Trickle

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

join|trickle()

返回:TrickleNode

Union

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

join|union(node ...Node)

返回:UnionNode

Where

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

join|where(expression ast.LambdaNode)

返回:WhereNode

Window

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

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

join|window()

返回:WindowNode


此页面是否有帮助?

感谢您的反馈!


Flux 的未来

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

阅读更多

现已全面上市

InfluxDB 3 Core 和 Enterprise

快速启动。更快扩展。

获取更新

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

有关更多信息,请查看