Taking historical data from SQL and write to influxdb bucket

Fluxlang
,
22

Hi @grant1

So I change your flow little. (it not sure best way)
I can able to work with Node-red.

var timen = Date.now()
var n =0
msg.payload = msg.payload.map(function (value) {
    return {
        measurement: "test",
        fields: {
            PJ: value.PJ,
            task: value.task
        },
        tags: {
            loc: "ABC",
            unitnm: "15"
        }, 
        timestamp: Date.now() + n++
    }
});
return msg;

Thanks again @grant1 @scott .

23

@Yutaka
Does this mean you have your historical SQL data now written into InfluxDB with the correct timestamps?

24

@grant1
Strictly, it’s not the right time.
but current my purpose, can accept it if time shift within ms.

25

@Yutaka Awesome. I’m glad it works. The timestamps only have to be 1 nanosecond different for InfluxDB to recognize them as distinct points, so you can us 1ns in date.scale():

|> map(fn: (r) => ({r with _time: date.add(d: date.scale(d: 1ns, n: r.index), to: r._time)}))
1 Like
26

@scott
Hello again. I’m so happy to using above method.
so now, If I wish to transform one of field to tag(my case “PJ”), is the following flux will be fine?
my result looks ok. I wish to confirm it, if you have any suggestion.

import "date"
import "sql"

sql.from(
    driverName: "mysql",
    dataSourceName: "usradmin:passwd@tcp(localhost:3306)/prod",
    query: "SELECT PJ AS PJ,task AS task FROM prod.linkdata",
)
    |> map(fn: (r) => ({r with _time: now(), _measurement: "test"}))
    |> group(columns: ["_measurement"])
    |> map(fn: (r) => ({r with index: 1}))
    |> cumulativeSum(columns: ["index"])
    |> map(fn: (r) => ({r with _time: date.add(d: date.scale(d: 1ns, n: r.index), to: r._time)}))
    |> to(bucket: "mariadbtest",tagColumns: ["PJ"],fieldFn: (r) => ({"task": r.task,"index": r.index}))
27

@Yutaka, looks good to me. My only concern would be on subsequent runs of this query, you’d get duplicate index values because it recalculates the index from 1.

28

Dear @scott.

I understand, that index for date.scale. I will treat nicely at fieldFn.

Thank you very much about it.