代码之家 › 专栏 › 技术社区 › Igor Toropov

Kapacitor Lambda均值

kapacitor influxdb

0

Igor Toropov · 技术社区 · 7 年前

我试图用Kapacitor batch | query生成基线,方法是查询1、2、3和4周前InfluxDB的相同间隔,然后将其向前移动并按如下方式连接在一起:

var w1 = batch
    |query('SELECT mean(""value"") FROM ""MyDB"".""autogen"".""MetricName""')
        .offset(1w).period(period).every(1m).align().groupBy(time(1m))
    |shift(1w)

var w2 = batch
    |query('SELECT mean(""value"") FROM ""MyDB"".""autogen"".""MetricName""')
        .offset(2w).period(period).every(1m).align().groupBy(time(1m))
    |shift(2w)

var w3 = batch
    |query('SELECT mean(""value"") FROM ""MyDB"".""autogen"".""MetricName""')
        .offset(3w).period(period).every(1m).align().groupBy(time(1m))
    |shift(3w)

var w4 = batch
    |query('SELECT mean(""value"") FROM ""MyDB"".""autogen"".""MetricName""')
        .offset(4w).period(period).every(1m).align().groupBy(time(1m))
    |shift(4w)

var bj = w1
    |join(w2, w3, w3)
        .as('w1', 'w2', 'w3', 'w4')
        .fill('null')

var b = bj
    |eval(lambda: (""w1.mean"" + ""w2.mean"" + ""w3.mean"" + ""w4.mean"") / float(4.0))
        .as('avg')

然而,lambda似乎不支持Mean()或任何此类数学函数。它似乎也不支持检查空值。

有没有这样的方法来计算基线?

此外,一旦计算出基线,如何能够保持缓存,以便根据基线检查传入流数据?

感谢您的帮助!非常感谢。

3 回复 | 直到 7 年前

1

0

arturo mondelo 7 年前

首先,尝试在批处理变量上使用偏移量,而不是shift。

偏移量将取x之前的最小值、小时值、天值。。。

shift节点应用于连接过程,例如:

previous
    |shift(1w)
    |join(current)
    ......

这里有一个例子: https://github.com/influxdata/kapacitor/issues/746

关于用不同的时间连接4个不同的流,由于我之前的评论,我想它不会起作用……也许用union代替join node可以起作用,但不确定!

2

0

Alexey Markov 6 年前

“因为有些星期可能缺少一个值” -join将永远等待该值,不会发出其他相应的批处理,并导致内存泄漏。

您希望将其拆分为多个脚本,例如,计算所有4个句点的脚本,在每个句点上添加一个标记,如

|default().tag('stream', 'w1')

然后把它们发送到

|kapacitorLoopback()

3

0

Igor Toropov 5 年前

最后,我使用C#.NET内核创建了一个UDF,作为一个进程并排运行,查询InfluxDB,计算并缓存结果。