Skript zum Downsamplen einer Influx 2.x DB

[A Former User]

@manrum1 sagte in Skript zum Downsamplen einer Influx 2.x DB:

Danke im Voraus

Moin,

ob es passt, oder ob es gut gelöst ist, lassen wir mal dahingestellt

import "timezone"

option location = timezone.location(name: "Europe/Berlin")
option task = {name: "Downsampling Vb_Stromzaehler", cron: "15 0 * * *"}

data =
    from(bucket: "iobroker_strom")
        |> range(start: -1y, stop: now())
        |> filter(fn: (r) => r["_measurement"] == "sonoff.0.DVES_8AA766.SENSOR.SML.total_kwh")
        |> filter(fn: (r) => r["_field"] == "value")
        // Spalten "_start", "_stop", "ack", "from", "q", ausschliessen
        |> drop(columns: ["ack", "q", "from"])

data
    |> aggregateWindow(every: 1d, fn: last, timeSrc: "_time")
    // In Wh ohne Komma
    |> toInt()
    |> set(key: "_measurement", value: "Hauptzaehler")
    // Use the to() function to validate that the results look correct. This is optional.
    |> to(bucket: "Stromverbrauch", org: "iobroker_strom")

VG
Bernd

[manrum1]

@dp20eic said in Skript zum Downsamplen einer Influx 2.x DB:

|> set(key: "_measurement", value: "Hauptzaehler")

Super! Danke, habe wie folgt angepasst und es funktioniert!

import "date"

option task = {name: "0_move_tmp", every: 1y}

fullHourTime = date.truncate(t: now(), unit: 1h)

from(bucket: "iobroker-data")
    |> range(start: 2023-01-01T00:00:00Z, stop: fullHourTime)
    |> filter(fn: (r) => r["_measurement"] == "0_userdata.0.Strom.hb.Status.hichiHB.SM.Energy_Out")
    |> filter(fn: (r) => r["_field"] == "value")
    |> filter(fn: (r) => r._value > 0)
    |> difference()
    |> aggregateWindow(every: 15m, fn: sum, createEmpty: true)
    |> set(key: "_measurement", value: "energy")
    |> to(bucket: "testdb", fieldFn: (r) => ({"exportedWh": r._value}))

[Marc Berg]

@manrum1

Oben hattest du noch geschreiben, dass du "mehrere fields exportedWh, importedWh" in einem Measurement haben willst. Gibt es denn noch einen weiteren Datenpunkt "importedWh"?

[manrum1]

@marc-berg Hi Marc,
danke, dass du dich meldest. Ja, aber dafür lege dann eine eigene Task an. Oder hast Du einen besseren Vorschlag?

[Marc Berg]

@manrum1

Ich weiß nicht, ob ich die Aufgabenstellung richtig verstanden habe, wenn ja, dann müsstest du "<importedWh>" durch dein MEasurement ersetzen, in welchem die Werte stehen:

import "date"

option task = {name: "0_move_tmp", every: 1y}

fullHourTime = date.truncate(t: now(), unit: 1h)

from(bucket: "iobroker-data")
    |> range(start: 2023-01-01T00:00:00Z, stop: fullHourTime)
    |> filter(fn: (r) => 
        (r["_measurement"] == "0_userdata.0.Strom.hb.Status.hichiHB.SM.Energy_Out" or r["_measurement"] == "<importedWh>")
        and r["_field"] == "value"
        and r._value > 0)
    |> difference()
    |> aggregateWindow(every: 15m, fn: sum, createEmpty: true)
    |> pivot(rowKey:["_time"], columnKey: ["_measurement"], valueColumn: "_value")
    |> to(bucket: "testdb", fieldFn: (r) => ({"exportedWh": r["0_userdata.0.Strom.hb.Status.hichiHB.SM.Energy_Out"], "importedWh": r["<importedWh>"]}))

Im Ergebnis hast du dann ein Measurement, in welchem beide Werte als Fields enthalten sind.

[manrum1]

@marc-berg Verstehe, wäre natürlich einfacher. Probiere ich morgen gleich aus!
Danke wieder einmal!!

[manrum1]

@marc-berg Die Angabe des measrurements energy hat noch gefehlt, schaut jetzt so aus und funktioniert wunderbar:

import "date"

option task = {name: "0_move_tmp2", every: 1y}

fullHourTime = date.truncate(t: now(), unit: 1h)

from(bucket: "iobroker-data")
    |> range(start: 2023-01-01T00:00:00Z, stop: fullHourTime)
    |> filter(
        fn: (r) =>
            (r["_measurement"] == "0_userdata.0.Strom.hb.Status.hichiHB.SM.Energy_Out"
                or
                r["_measurement"] == "0_userdata.0.Strom.hb.Status.hichiHB.SM.Energy_Total")
                and
                r["_field"] == "value" and r._value > 0,
    )
    |> difference()
    |> aggregateWindow(every: 15m, fn: sum, createEmpty: true)
    |> pivot(rowKey: ["_time"], columnKey: ["_measurement"], valueColumn: "_value")
    |> set(key: "_measurement", value: "energy")
    |> to(
        bucket: "testdb2",
        fieldFn: (r) =>
            ({
                "exportedWh": r["0_userdata.0.Strom.hb.Status.hichiHB.SM.Energy_Out"],
                "totalWh": r["0_userdata.0.Strom.hb.Status.hichiHB.SM.Energy_Total"],
            }),
    )

Allerdings habe ich jetzt noch das Problem meines Datenverlusts (https://forum.iobroker.net/topic/62903/gelöst-backitup-und-influxdbv2/79?_=1703005510202) und das betrifft ja leider alle Daten in meiner Datenbank. Beispiel:

Bei aufsummiereten Counter sind die Daten ja in Summe nach der Verlustwoche wieder da. Hast du eine Idee, wie ich diese als Durchschnittswerte wieder auf die Vortage verteilen kann?

[Marc Berg]

@manrum1 sagte in Skript zum Downsamplen einer Influx 2.x DB:

Hast du eine Idee, wie ich diese als Durchschnittswerte wieder auf die Vortage verteilen kann?

Nicht schön, aber pragmatisch:

1. Export einer "Vorlage", den Range am besten so anpassen, dass du ein oder mehrere Werte vor und nach der Lücke erwischst:

influx query 'from(bucket:"iobroker") |> range(start: -1d)   |> filter(fn: (r) => r["_measurement"] == "0_userdata.0.Strom.hb.Status.hichiHB.SM.Energy_Out")
  |> filter(fn: (r) => r["_field"] == "value")' --raw > outputcsv.txt

2. Dann hast du ein CSV ähnliches Format, in welchem du einfach Zeilen kopierst, wieder neu einfügst und die Spalten "_time" und "_value" nach Belieben anpasst und z.B. einen Wert pro Tag einfügst. Der Header muss unverändert bleiben.

3. Import der angepassten Datei über die InfluxDB WEBgui (File Upload, CSV)

EDIT: mit diesem Verfahren lassen sich bestehende Werte ändern und neue hinzufügen, aber keine Werte löschen!

EDIT2: Export korrigiert

[manrum1]

@marc-berg So klappt der Import leider nicht.

Habe Header so belassen:

Result: _result
Table: keys: [_start, _stop, _field, _measurement]
                   _start:time                      _stop:time           _field:string                                   _measurement:string                      _time:time                  _value:float
------------------------------  ------------------------------  ----------------------  ----------------------------------------------------  ------------------------------  ----------------------------
2023-11-26T23:00:00.000000000Z  2023-12-04T01:00:00.000000000Z                   value  0_userdata.0.Strom.hb.Status.hichiHB.SM.Energy_Total  2023-11-26T23:00:44.812000000Z                    16407.2868
2023-11-26T23:00:00.000000000Z  2023-12-04T01:00:00.000000000Z                   value  0_userdata.0.Strom.hb.Status.hichiHB.SM.Energy_Total  2023-11-26T23:00:46.806000000Z                    16407.3246

[Marc Berg]

@manrum1 sagte in Skript zum Downsamplen einer Influx 2.x DB:

So klappt der Import leider nicht.

sorry, habe ich schon länger nicht gemacht. Du musst den Parameter "--raw" beim Export mitgeben. Dann sieht der Aufbau der Datei anders aus. Der Rest bleibt aber gleich.

#group,false,false,true,true,false,false,true,true
#datatype,string,long,dateTime:RFC3339,dateTime:RFC3339,dateTime:RFC3339,double,string,string
#default,_result,,,,,,,
,result,table,_start,_stop,_time,_value,_field,_measurement
,,1,2023-12-20T11:50:19.573820995Z,2023-12-20T12:50:19.573820995Z,2023-12-20T11:54:05.709Z,6.8,value,mqtt.0.Lora.esp03.abshum

[manrum1]

@marc-berg Ja, so geht's. Werde mal die wichtigsten Daten mit dieser Methode "auffüllen". Wird ne Heidenarbeit.

Danke!