InfluxDB 2.x - Grundlagen für Einsteiger

Eine sehr beliebte Datenbank im Smart Home Umfeld ist die Time Series Datenbank InfluxDB. Diese ist perfekt dafür geeigent, um große Datenmengen zu sammeln, zu aggregieren und sehr schnell abzufragen. Viele verwenden diese Datenbank bereits sehr lange (so wie ich auch), aber haben sich nie ernsthaft mit den Grundlagen beschäftigt. Und genau diese Wissenslücke möchte ich heute mit diesem Beitrag und dem zugehörigen Video schließen!

Was Du benötigst?

• Eine bestehende InfluxDB 2.x Installation (Im Video wird 2.6.1 verwendet)
• Etwas Zeit

Video

Daten per HTTP-API schreiben

Um Daten schreiben zu können, brauchst Du folgende Informationen:

• Bucket
• Organization
• Token

curl -X POST \
--data 'wetter,stadt=hamburg,land=de temperatur=7' \
-H 'Authorization: Token FlL3zaSngsvDWRTvYWneRkrytB4XJm8-MQPBLJ9TtPIOUg9y80YOSvZWpBk-kugJLnj-deSTKFfrYWVyXNTWww==' \
-H 'Content-Type: text/plain' \
'http://172.16.0.251:8086/api/v2/write?bucket=test&org=hausautomation'

Und schon bist Du bereit, um Informationen im Line Protocol in die Datenbank zu schreiben! Das schöne ist: HTTP spricht so gut wie jeder Client. Selbst mit einem Gira X1 o.ä. kann man so relativ einfach neue Daten in die Datenbank schreiben. Man merkt förmlich, dass hier die Anforderungen an den Client möglichst gering gehalten wurden.

Dein Token kannst Du z.B. über das CLI abfragen:

influx auth list --user mkleine --hide-headers | cut -f 3

Demo-Daten aus dem Video

energie,quelle=ehz zaehlerstand=10000,preis=0.30 1674428400000000000
energie,quelle=ehz zaehlerstand=10100,preis=0.30 1674430200000000000
energie,quelle=ehz zaehlerstand=10300,preis=0.30 1674431100000000000
energie,quelle=ehz zaehlerstand=10500,preis=0.35 1674432000000000000
energie,quelle=ehz zaehlerstand=10900,preis=0.35 1674432900000000000

Entwicklung des Zählerstandes (ohne Preise):

from(bucket: "test")
  |> range(start: 2023-01-22T23:00:00.000Z, stop: 2023-01-23T02:00:00.000Z)
  |> filter(fn: (r) => r._measurement == "energie")
  |> filter(fn: (r) => r._field == "zaehlerstand")
  |> difference()
  |> aggregateWindow(every: 15m, fn: last, createEmpty: false)
  |> yield(name: "last")

Preis errechnen (fehlerhafte Version mit Gedankenfehler):

from(bucket: "test")
  |> range(start: 2023-01-22T23:00:00.000Z, stop: 2023-01-23T02:00:00.000Z)
  |> filter(fn: (r) => r._measurement == "energie")
  |> filter(fn: (r) => r._field == "zaehlerstand" or r._field == "preis")
  |> aggregateWindow(every: 15m, fn: last, createEmpty: false)
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> map(fn: (r) => ({r with _value: r.zaehlerstand / 1000.0 * r.preis}))
  |> difference()

Preis errechnen (richtig):

from(bucket: "test")
  |> range(start: 2023-01-22T23:00:00.000Z, stop: 2023-01-23T02:00:00.000Z)
  |> filter(fn: (r) => r._measurement == "energie")
  |> filter(fn: (r) => r._field == "zaehlerstand" or r._field == "preis")
  |> aggregateWindow(every: 15m, fn: last, createEmpty: false)
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> difference(columns: ["zaehlerstand"])
  |> map(fn: (r) => ({r with _value: r.zaehlerstand / 1000.0 * r.preis}))

Tasks

Mit folgendem Task aggregiere ich die Daten regelmäßig und speichere sie in ein separates Bucket.

from(bucket: "test")
  |> range(start: -24h)
  |> filter(fn: (r) => r._measurement == "energie")
  |> filter(fn: (r) => r._field == "zaehlerstand" or r._field == "preis")
  |> aggregateWindow(every: 15m, fn: last, createEmpty: false)
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> difference(columns: ["zaehlerstand"])
  |> map(fn: (r) => ({r with _value: r.zaehlerstand / 1000.0 * r.preis}))

  |> to(
    bucket: "test-langzeit",
    tagColumns: ["quelle"],
    fieldFn: (r) => ({ "preis": r._value, "verbrauch": r.zaehlerstand })
  )

Daten löschen

Ein ganzes Bucket lässt sich ganz einfach über die Weboberfläche der InfluxDB löschen. Wenn Du ein einzelnes Measument löschen möchtest, kannst Du das ebenfalls über die HTTP API tun:

curl -X POST \
--data '{"predicate": "_measurement=\"energy-stats\"", "start": "1970-01-01T00:00:00Z", "stop": "2099-12-31T23:59:59Z"}' \
-H 'Authorization: Token FlL3zaSngsvDWRT...' \
-H 'Content-Type: application/json' \
'http://172.16.0.251:8086/api/v2/delete?bucket=test&org=hausautomation'