1

在将它们放入 influxdb 之前，我想用 kapacitor 汇总来自 telegraf 的最后一分钟的数据，而且我也需要计算几个百分位数。所以我为测试写了一个简单的勾号

var firstPerc = stream
    |from()
        .measurement('my_tmp_measurement_from_telegraf')
var secondPerc = stream
    |from()
        .measurement('my_tmp_measurement_from_telegraf')
firstPerc
    |join(secondPerc)
        .as('fp', 'sp')
        |percentile('fp.myAggVal', 50.0)
        |eval(lambda: "percentile")
            .as('50p')
        |percentile('sp.myAggVal', 90.0)
        |eval(lambda: "percentile")
            .as('90p')
        |window()
            .period(60s)
            .every(60s)
            .align()
        |influxDBOut()
            .database('myDBInInflux')
            .retentionPolicy('autogen')

在我的数据库中，我只有第 50 个百分位数的值，我对此并不感到惊讶，因为我在评估中使用了“百分位数”，但我仍然无法在 Kapacitor 文档中找到任何关于如何获得所需结果的线索。

在这里，您有我渴望的“视觉”结果：

time 50p 90p someOtherP's otherDataICanPropablyHandle

哈！

4

1 回答 1

1

您两次使用相同的测量流（以及其中的相同数据），因此会弹出数据。首先，您应该保存测量流：

var myStream = stream
    |from()
        .measurement('my_tmp_measurement_from_telegraf')

接下来使用保存的测量定义流。您应该在这里定义适当的分组、评估等：

var firstPerc = myStream
    |percentile('myAggVal', 50.0)
    |eval(lambda: "percentile")
        .as('percentile')
    |window()
        .period(60s)
        .every(60s)
        .align()
var secondPerc = myStream
    |percentile('myAggVal', 90.0)
    |eval(lambda: "percentile")
        .as('percentile')
    |window()
        .period(60s)
        .every(60s)
        .align()

最后，是时候定义加入流了：

var joinedStreams = firstPerc
    |join(secondPerc)
        .as('50', '90')
        .tolerance(1s)
        .streamName('measurementName')
        |influxDBOut()
            .database('myDBInInflux')
            .retentionPolicy('autogen')
            .create()

输出：

time                50.percentile 90.percentile

我强烈建议使用 .tolerance()，它将在相同的容差期内对测量进行分组。

于 2017-11-22T14:01:33.837 回答