UPDATE 31.10.: Amazon Alexa - ioBroker Skill läuft aus ?

[mickym]

@spacerx sagte in Datenaufzeichnung Retention InfluxDB 2.0:

Wie könnte so ein Task in Code aussehen? Ich nehme mal an da ich Flux nutze dann auch in Flux.

Schau Dir mal - da gibts Beispiele:

https://awesome.influxdata.com/docs/part-3/tasks/?utm_source=influx&utm_medium=referral&utm_campaign=2022_training_influxdb-university_global&utm_content=101-influxdb-essentials#a-downsampling-task

Das sind Downsampling Tasks - man kann in der ui ein Task erstellen - ich mache in den University ein paar von diesen Online Kurse. Im Prinzip machst Du mit aggregate Windows - konsolidierte Tabellen die Du dann in das nächste Bucket schreibst. In dem Beispiel wird über eine Minute der Mittelwert gebildet - geht natürlich auch über 1 Stunde oder 1 Tag etc.. Schlüssel ist die aggregateWindow Funktion.

[mickym]

@spacerx sagte in Datenaufzeichnung Retention InfluxDB 2.0:

Da mich zum Beispiel 2 Jahre später noch der wöchentliche Verbrauch in der KW 22 von vor 2 Jahren interessieren könnte, lasse ich zum 1ten des Folgemonats nur die Zählerstände der Wochen des Monats in das Bucket "Stromverbrauch" schreiben. Oder vielleicht doch schon Differenz errechnen?

Mit last kann man den letzten Zählerstand als aggregierten Zustand des jeweiligen Zeitfensters verwenden. Man kann sicher auch die Differenz aber da bin ich noch nicht so weit - ich lerne noch.

[SpacerX]

@mickym ich bin schon fleißig am einlesen und Probiere das einfach mal aus.

Für mein Beispiel würde ja täglich voll ausreichen.

option task = {name: "Downsampling Stromzaehler", every: 1h}

  data = from(bucket: "ioBroker")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "Vb_Stromzaehler")
  |> filter(fn: (r) => r["from"] == "system.adapter.javascript.0")
  
  data
  |> aggregateWindow(every: 1d, fn: last, timeSrc: "_start")
  |> difference()
  |> map(fn: (r) => ({r with _value: r._value / 1000.0}))
  |> to(bucket: "Stromverbrauch", org: "RBE")

Das würde jetzt immer den letzten Wert zur vollen Stunde in das neue Bucket schreiben. Richtig?

Würde es hier Sinn machen das:

import "timezone" // import wegen der Berechnung um 02:00:00
option location = timezone.location(name: "Europe/Berlin")

mit einzufügen? Bei der aggregieren und difference macht das schon was aus.

[mickym]

@spacerx sagte in Datenaufzeichnung Retention InfluxDB 2.0:

Das würde jetzt immer den letzten Wert zur vollen Stunde in das neue Bucket schreiben. Richtig?

Alle Tage - für alle Stunden müsstest es

aggregateWindow(every: 1h ..

heißen

[mickym]

@spacerx sagte in Datenaufzeichnung Retention InfluxDB 2.0:

mit einzufügen? Bei der aggregieren und difference macht das schon was aus.

Der nimmt aber local Timezone - aber abgespeichert werden alle Zeiten glaub in UTC.

[SpacerX]

@mickym Das eigentliche Downsampling bekomme ich hin, nur Stimmen die Daten dann einfach nicht mehr. Schuld daran ist das ganze geranze mit den Timezone bezogenen Berechnungen.

Ich hab das jetzt gefühlte 50 mal in allen Varianten durchexerziert.
Ich mache das aggregieren auf den Tag in Zeitzohne Berlin dann Stimmen die Werte.
Lasse ich das nun Downsampeln werden die Werte nicht in Timezone Berlin berechnet sondern +3 Stunde versetzt.
Korrigiere ich das beim anzeigenden Query mit Timezone Berlin komme ich auf +1 Stunde raus.
Ich raffe das nicht wie intern gerechnet wird.

Wo kann man die Local Timezone in InfluxDB sehen einstellen? Vielleicht auch die Timezone einstellen?

[ftd]

@spacerx By default, Flux uses the UTC time zone.

Korrigiere ich das beim anzeigenden Query mit Timezone Berlin komme ich auf +1 Stunde raus.

Mach doch einfach noch ein

option location = timezone.fixed(offset: -1h)

dahinter?!

[SpacerX]

@ftd Der Vollständigkeit halber noch der Erfolg bringende Task.

import "timezone"

option location = timezone.location(name: "Europe/Berlin")

option task = {name: "Downsampling Vb_Stromzaehler", cron: "1 0 * * *"}

data =
    from(bucket: "ioBroker")
        |> range(start: -task.every)
        |> filter(fn: (r) => r["_measurement"] == "Vb_Stromzaehler")
        |> filter(fn: (r) => r["from"] == "system.adapter.javascript.0")

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

und noch der Query hinterher, für monatlich übers Jahr.

import "timezone" // import wegen der Berechnung um 02:00:00
option location = timezone.location(name: "Europe/Berlin")
from(bucket: "Stromverbrauch")
  |> range(start: -1y, stop: now())
  |> filter(fn: (r) => r["_measurement"] == "Hauptzaehler")
  |> aggregateWindow(every: 1mo , fn: last)
  |> difference()

Ich habe nun aber auch etwa 10 Messstellen und nach 3-4 andere Measurements die ich etwas länger behalten möchte. das wird Fleißarbeit.

[SpacerX]

Ich schiebe noch paar Überlegungen nach:

• Downsampling um 00:15 Uhr Nachts. Einmal am Tag reicht denke ich die aktuellen Daten für die Visu bekomme ich ja aus dem Quellbucket.

• Gleich auch die Spalten "ack", "q", "from" mit entfernen. Ich glaube nicht das man die jemals wieder braucht.

import "timezone"

option task = {name: "Downsampling Vb_Stromzaehler", cron: "15 0 * * *"}

option location = timezone.location(name: "Europe/Berlin")

data =
    from(bucket: "ioBroker")
        |> range(start: -2mo, stop: now())
        |> filter(fn: (r) => r["_measurement"] == "Vb_Stromzaehler")
        |> filter(fn: (r) => r["from"] == "system.adapter.javascript.0")
        // Spalten "_start", "_stop", "ack", "from", "q", ausschliessen
        |> drop(columns: ["ack", "q", "from"])

data
    |> aggregateWindow(every: 1d, fn: last, timeSrc: "_start")
    // 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: "RBE")

Wichtig ist das der drop Befehl in dem oberen data Block steht, ansonsten wird ein Fehler angemeckert.

Die beste und korrekte Darstellung in Grafana ergibt sich mit:

import "timezone" // import wegen der Berechnung um 02:00:00
option location = timezone.location(name: "Europe/Berlin")
from(bucket: "Stromverbrauch")
  |> range(start: -1y, stop: now())
  |> filter(fn: (r) => r["_measurement"] == "Hauptzaehler")
  |> aggregateWindow(every: 1mo , fn: last,  timeSrc: "_start") // ,createEmpty: false )
  |> difference()

Die Anpassung der X-Bezeichnung mit Override

Vieleicht sucht das ja mal jemand und kann es hier finden.:blush:

[bommel_030]

@spacerx
Ans Downsampling hab ich mich noch nicht rangetraut, das hebe ich mir für den Urlaub auf ;-)
Danke aber schonmal für die Grafana-Darstellung, läuft einwandfrei.
Hast du dir zufällig mal Gedanken darüber gemacht den Verbrauch des Vorjahres in der selben Darstellung mit aufzunehmen? Also 12 Balken aus dem Vorjahr im Hintergrund und das aktuelle davor?

[SpacerX]

@bommel_030 Dahinter wird schwierig. Daneben sollte gehen.

[bz7i]

@SpacerX: Danke für den super Tipp. Inzwischen hab ich mich etwas in Flux eingearbeitet :-)

Da ich auch Probleme mit nicht passenden Werten hatte, kann Ich ja mal zeigen wie Ich es gelöst hab.

fn: last bringt folgendes Problem. Der Stromverbrauch wird nicht von 0:00 - 0:00Uhr berechnet. Sondern mit dem Wert zuvor. Der letzte also. z.B. 23:00 - 23:00Uhr. Finde Ich nicht ganz so schön. Deshalb hab ich fn: first verwendet.

Damit passten die Zeiten aber nicht mehr. Das hab ich folgt gelöst:

 |> map(fn: (r) => ({r with _time: date.add(d: -1h, to: r._time)}))

mit d: -1h, hab ich den Zeitstempel um eine Stunde verschoben.

Mein kompletter Code um auf Stundenbasis zu Downsampeln sieht wie folgt aus :

import "date"

option task = {name: "Stromverbrauch", every: 1h}

data =
    from(bucket: "ioBroker")
        |> range(start: -24h, stop: 1h)
        |> filter(fn: (r) => r["_measurement"] == "javascript.0.Info.StromZaehler.Zaehlerstand_Min")
        |> filter(fn: (r) => r["_field"] == "value")

data
    |> aggregateWindow(every: 1h, fn: first, createEmpty: false)
    |> difference(nonNegative: true, columns: ["_value"])
    |> toInt()
    |> map(fn: (r) => ({r with _value: r._value / 10}))
    |> map(fn: (r) => ({r with _time: date.add(d: -1h, to: r._time)}))
    |> set(key: "_Interval", value: "1h")
    |> to(bucket: "Energie", org: "inFluxDB", tagColumns: ["_Interval"])

Hoffe ich konnte damit helfen....

[cainam]

kleine Frage meinerseits. Könnte man so einen Task nicht auch in das gleiche Bucket zurückschreiben.

Dh wenn ich zB Daten die älter als 90Tage filtere und diese aggregiere (zB Tagesdurchschnitt) und diese dann wieder ins gleiche Bucket schreibe...

habe ich dann nicht von ganz alten daten bis heute -90Tage ein Tagesdurchschnitt und -89 Tage bis heute die Minutenaufzeichnung?

evt müsste man ein zwischenspeichern machen?

option task = {
  name: "AggregationTask",
  every: 24h,
  offset: 0h
}

from(bucket: "iobroker")
  |> range(start: -2y, stop: -90d)
    |> filter(fn: (r) => r["_measurement"] == "Aussentemperatur")
    |> filter(fn: (r) => r["_field"] == "value")
  |> aggregateWindow(every: 1d, fn: mean, createEmpty: true)
  |> to(bucket: "temporary_bucket")

from(bucket: "iobroker")
  |> range(start: -2y, stop: -90d)
  |> drop()             

from(bucket: "temporary_bucket")
  |> range(start: -2y, stop: -90d)
  |> to(bucket: "iobroker")

jrgendwie so, jedoch geht das oben so nicht beim versucht

[Marc Berg]

@cainam
Du kannst die Daten natürlich in das gleiche Bucket schreiben, das Problem (dass du ja offensichtlich schon gefunden hast) besteht im Löschen der Daten. Das geht meines Wissens aus einem Task heraus nicht, sondern nur per API oder CLI. Theoretisch könnte man das Löschen über einen Cronjob durchführen und die Aggregate Funktionen inkl. Schreiben über ein temporäres Bucket per Task und das Ganze zeitlich synchronisieren ... Schwierig ...

Ich glaube, für diese Anforderung sind die Tasks raus. Stattdessen würde ich das ganze über ein Bash-Skript lösen.

[cainam]

Bist du sicher geht das mit drop nicht?

Es wäre even einfach die Daten so zu reduzieren anstatt alles neue buckets zu machen… auch vergessen Risiko ist massiv tiefer

[Marc Berg]

@cainam sagte in Datenaufzeichnung Retention InfluxDB 2.0:

Bist du sicher geht das mit drop nicht?

Ja.

[bitwicht]

@mickym

Re: Datenaufzeichnung Retention InfluxDB 2.0

Dazu hätte ich noch eine Frage.

Das Bucket "iobroker" stelle ich auf 180 Tage.
Das Bucket "iobroker-downsampling" stelle ich auf 5 Jahre.

Ich schreibe jetzt monatlich den Wasserverbrauch weg.

Wenn ich nun mit Grafana eine Auswertung über 3 Jahre machen, müsste ein Teil der Daten aus dem iobroker Bucket und der andere Teil aus dem iobroker-downsampling Bucket kommen.

Wäre das so der richtige Aufbau?

Würde ich das dann so umbauen müssen:

from(bucket: "iobroker" & "iobroker-downsampling")
  |> range(start: -12mo)
  |> filter(fn: (r) => r["_measurement"] == "Wasserzaehler-CAM-Haus")
  |> filter(fn: (r) => r["_field"] == "value")
  |> difference()
  |> aggregateWindow(every: 1mo, fn: sum, createEmpty: false, timeSrc: "_start")
  |> keep(columns: ["_value", "_time"])

[Marc Berg]

@bitwicht sagte in Datenaufzeichnung Retention InfluxDB 2.0:

Würde ich das dann so umbauen müssen:

So (mit "&") geht das nicht. Dafür sollte die union Funktion geeignet sein. Du machst zwei Abfragen auf jeweils ein Bucket und "klebst" die mit union sozusagen untereinander.

https://docs.influxdata.com/flux/v0/stdlib/universe/union/

[bitwicht]

@Marc-Berg

Das macht natürlich alles viel komplexer.

Die Idee von @SpacerX ein Bucker "forever" und ein Bucket 180 Tage finde ich ja viel scharmanter.
Wäre dann aber nicht "best practice"

[Marc Berg]

@bitwicht sagte in Datenaufzeichnung Retention InfluxDB 2.0:

Das macht natürlich alles viel komplexer.

Einfach kann jeder...

Wäre dann aber nicht "best practice"

Best practise gibt es hier m.E nicht. Dazu sind die Anforderungen, Daten(mengen) und das eigene Wissen einfach zu unterschiedlich. Man muss sich den Weg suchen, der einem passt und der einem nicht die Hälfte der Lebenszeit wegfrisst.