Tasks could work manually, but not automatically?

InfluxDB 2 Tasks
, , ,
1

Hi,
I have met a problem. when I run task manually, would insert data to “control” bucket. But no data insert into “control” bucket when run task automatically, also logs record as sucessfully. Below is my task , can anyone help?

option task = { 

  name: "control_power_forever",

  every: 15m,

}

control = from(bucket: "forever")

    |> range(start: -task.every)

    |> filter(fn: (r) => r._measurement == "T3_5s")

    |> filter(fn: (r) =>  r["_field"] == "control_power" and r.type =="2" )

    |> aggregateWindow(every: 15m, fn: last)

    |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")

    |> group(columns: ["_time"], mode:"by")

    |> unique(column: "control_power")

  

original = () => {

    meter = from(bucket: "forever")

        |> range(start: -task.every)

        |> filter(fn: (r) => r._measurement == "T3_5s")

        |> filter(fn: (r) => r["_field"] == "ac_history_negative_power_in_kwh" and r.type =="2" )

        |> aggregateWindow(every: 15m, fn: last)

        |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")

        |> difference(columns:["ac_history_negative_power_in_kwh"])

        |> group(columns: ["_time"], mode:"by")

        |> map(fn: (r) => ({ r with ac_history_negative_power_in_kwh: r.ac_history_negative_power_in_kwh * 4.0 }))

        |> sum(column:"ac_history_negative_power_in_kwh")

   ems = from(bucket: "forever")

        |> range(start: -task.every)

        |> filter(fn: (r) => r._measurement == "T1_5s")

        |> filter(fn: (r) => r["_field"] == "ems_history_output_energy" )

        |> aggregateWindow(every: 15m, fn: last)

        |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")

        |> difference(columns:["ems_history_output_energy"])

        |> group(columns: ["_time"], mode:"by")

        |> map(fn: (r) => ({ r with ems_history_output_energy: r.ems_history_output_energy * 4.0 }))

        |> sum(column:"ems_history_output_energy")

    return join( tables: {meter, ems},on: ["_time"],)

        |> group(columns: ["_time"], mode:"by")

        |> map(fn: (r) => ({ r with originalDemand: r.ems_history_output_energy + r.ac_history_negative_power_in_kwh}))

}

originalDemand = original()

join( tables: {originalDemand, control}, on: ["_time"], )

 |> map(fn: (r) => ({r with max : if ( r.control_power > r.ac_history_negative_power_in_kwh) then (r.originalDemand - r.control_power) else (r.originalDemand - r.ac_history_negative_power_in_kwh)}))

 |> map(fn: (r) => ({r with _value : if r.max > 0 then r.max  else 0.0}))

 |> map(fn: (r) => ({r with _field : "real"}))

 |> drop(columns: ["ammeterId","uuid"])

 |> to(bucket: "control")
2

Hi @Jacobsz,
This might be down to how you are pulling the range of data. Could you try this instead:

import "influxdata/influxdb/tasks"
#######
 |> range(start: tasks.lastSuccess(orTime: -task.every))

Make sure to also include the import of the tasks library.

4

Hi jay,
thanks for your help, also it still doesn’t work, but it really help me to find the root cause. As you mentioned, it really caused by the range of data. I used some complex funtions, when run the task automatically, some data may not be generated. So there would be no data output to destinated bucket，and it looks like the task doesn’t work。Now I increase the task inteval，and It works.
Could you help with another issue. I want to aggregate some data with function differcece from 8:00 -9:00AM in everyday from May 1st to May 12th. How can I write the influx sql?

1 Like
5

Hi @Jacobsz,
Sorry for the late reply. You could do something like this:

import "date"
from(bucket: "northbound")
  |> range(start: -7d)
  |> filter(fn: (r) => r["_measurement"] == "genData")
  |> filter(fn: (r) => r["_field"] == "fuel")
  |> filter(fn: (r) => r["host"] == "influxdata-roadshow")
  |> filter(fn: (r) => date.hour(t: r["_time"]) > 9 and date.hour(t: r["_time"]) < 17)
  |> yield(name: "sub-range")

date.hour pulls out the hour from each row and then you can filter against that. So in this example we are only including results from 9 - 5 :slight_smile:

1 Like
6

Hi，
When I use function difference (), one row will caculate the diff between 7PM yesterday and 5AM today. I onlyt expect the difference result in everyday itself only.

7

Hi @Jacobsz,
You could use a window to group by every 24 hours after the filter. That way each day’s data it kept within its own table. You can then perform the difference. We also found a better way of doing the subrange for you:

subrange = from(bucket: "northbound")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "genData")
  |> filter(fn: (r) => r["_field"] == "fuel")
  |> filter(fn: (r) => r["host"] == "influxdata-roadshow")
  |> hourSelection(start: 9, stop: 16, timeColumn: "_time",)
 // |> yield(name: "subrange")

subrange 
  |> window(every: 24h)
  |> difference()
  |> yield(name: "differnce")

Note that hourSelection will return every result up until the last record in range. i.e in my case since i selected 16 → 16:59:59:59 etc.

1 Like
8

H Jay,
Sorry to reply late. It works for me. I sincerely feel grateful to you .

1 Like
9

No worries at all! Happy to help anytime :slight_smile: