InfluxDB

From Torben's Wiki

Install on Raspberry Pi in Raspbian

Nice Howto: [1] 

wget -qO- https://repos.influxdata.com/influxdb.key | sudo apt-key add -
source /etc/os-release
test $VERSION_ID = "7" && echo "deb https://repos.influxdata.com/debian wheezy stable" | sudo tee /etc/apt/sources.list.d/influxdb.list
test $VERSION_ID = "8" && echo "deb https://repos.influxdata.com/debian jessie stable" | sudo tee /etc/apt/sources.list.d/influxdb.list
test $VERSION_ID = "9" && echo "deb https://repos.influxdata.com/debian stretch stable" | sudo tee /etc/apt/sources.list.d/influxdb.list
sudo apt-get install influxdb
sudo service influxdb restart

influxdb.conf

in /etc/influxdb/influxdb.conf 

[http]
# Determines whether HTTP endpoint is enabled.
enabled = true
# require login of user
auth-enabled = true
# prevent logging of each HTTP request to reduce load on ssd
log-enabled = false

[logging]
level = "warn"

in ~/.bashrc 

export INFLUX_USERNAME=admin
export INFLUX_PASSWORD=password1

Runs per default on port 8086

Administration

CLI: command line interface

sudo apt-get install influxdb-client
influx -precision rfc3339 # for human readable time format
# or 
influx
precision rfc3339

User permissions

user permissions are defined per database, so if you need only one level of permissions, one database might be enough 

CREATE DATABASE raspi
USE raspi

suggestion: one user per permission 

create user uadmin with password 'password1' WITH ALL PRIVILEGES
create user uwrite with password 'password2'
create user uread  with password 'password3'
grant read on raspi to uread
grant write on raspi to uwrite

Backup and Restore

influxd backup  -portable mypath
influxd restore -portable mypath

Renaming a Database via creating a copy of its tables

Attention: this will convert tags to fields is group by * missing 

CREATE DATABASE mydb2
SELECT * INTO mydb2.autogen.table1 from mydb1.autogen.table1 group by *
DROP DATABASE mydb1

General commands

SHOW commands

SHOW DATABASES
SHOW MEASUREMENTS
SHOW RETENTION POLICIES
# display the field types of "table" myMeasurement
SHOW FIELD KEYS FROM myMeasurement
# show tags
SHOW TAG   KEYS FROM myMeasurement
# for example to be used in Grafana variables of type Query
SHOW TAG VALUES FROM myMeasurement WITH KEY =~ //
# CQ
SHOW CONTINUOUS QUERIES

SELECT statement

precision rfc3339
# for human readable timeformat
USE myDB
SELECT * FROM myMeasurement WHERE time > now() - 3d
SELECT * FROM myMeasurement WHERE myField > 12.2

# cast field into different type into new measurement
SELECT watt::float INTO myMeasurement2 FROM myMeasurement1

# copy into new measurement: 
SELECT watt::float, kWh_total_in::float, kWh_total_out::float INTO tasmota_MT681 FROM tmp
# use GROUP BY * to keep field types AND tags
SELECT watt_now, watt_last, kWh_total INTO tmp FROM Shelly GROUP BY *


Last value per day 

SELECT last("kWh_total_in") as kWh_total_in, last("kWh_total_out") as kWh_total_out
FROM "tasmota_MT681"
WHERE 1=1
-- AND time > now() -2d
AND time < '2024-07-13T00:00:00+02:00'
GROUP BY time(1d) fill(previous)

DELETE alternative

DELETE is not supported in Influx V1

alternative: 

# filter data into new tmp measurement and cast to float instead
# use GROUP BY * to keep field types and tags
SELECT watt::float, kWh_total_in, kWh_total_out INTO tmp FROM tasmota_MT681 WHERE kWh_total_in >0 GROUP BY *
# drop original measurement
DROP measurement tasmota_MT681
# restore from tmp
SELECT watt, kWh_total_in, kWh_total_out INTO tasmota_MT681 FROM tmp GROUP BY *
# cleanup
DROP MEASUREMENT tmp

Retention Policies

SHOW RETENTION POLICIES
# keep data for 10 years instead of 6 days (default)
CREATE RETENTION POLICY "years10" ON raspi DURATION 520w REPLICATION 1 DEFAULT
DROP   RETENTION POLICY "years10" ON raspi

# infinite 
CREATE RETENTION POLICY "inf" ON "raspi" DURATION INF REPLICATION 1

# if switching the default RETENTION POLICY old data will require a specific select
# altering the default RP:
ALTER RETENTION POLICY autogen ON raspi DURATION 90d SHARD DURATION 1d REPLICATION 1 DEFAULT
ALTER RETENTION POLICY autogen ON collectd DURATION 7d SHARD DURATION 1d REPLICATION 1 DEFAULT

Continuous Query / Auto aggregation

SHOW CONTINUOUS QUERIES

CREATE CONTINUOUS QUERY "cq_mt681_1d" ON "raspi" BEGIN \
 INTO "inf"."tasmota_MT681_day_end" \
 SELECT last("kWh_total_in") as kWh_total_in, last("kWh_total_out") as kWh_total_out \
 FROM "tasmota_MT681" \
GROUP BY time(1d) fill(previous) \
END

DROP CONTINUOUS QUERY "cq_mt681_1d" ON "raspi"

Connection via CURL

curl -G http://raspi3:8086/query -u uwrite:password2 --data-urlencode "q=SHOW DATABASES"

Insert via REST API and CURL

# precision=s (second) is important for performance, if higher accuracy is needed use ms, not ns which is the default!)
VALUE="123.123"
curl -i \
       -u uwrite:password2 \
       -XPOST "http://localhost:8086/write?db=raspi&precision=s" \
       --data-binary "myTable,myTag=SourceA myValue=$VALUE"

SELECT, INSERT and UPDATE via Python

SELECT into Pandas DataFrame

def read_data() -> pd.DataFrame:
    """Read data from Influx DB into DataFrame."""
    client = connect2_df(credentials_read)
    query = f'SELECT * FROM "{RETENTION}"."{MEASUREMENT}"'  # nosec  # noqa: S608
    result = client.query(query)
    df: pd.DataFrame = result[MEASUREMENT]  # type: ignore
    df.index.name = "time"
    client.close()
    return df

INSERT / UPDATE via Pandas DataFrame

def insert(df: pd.DataFrame) -> None:
    """Insert of df into Influx DB."""
    client = connect2_df(credentials_write)
    if len(df) > 0:
        client.write_points(
            df,
            MEASUREMENT,
            tag_columns=["room"],
            protocol="line",
            batch_size=1000,
            retention_policy=RETENTION,
            time_precision="s", # ns .. s
        )
    client.close()

INSERT JSON data via InfluxDBClient

see [2] 

from influxdb import InfluxDBClient
client = InfluxDBClient(host='192.168.178.31', port=8086, username='write', password='password2 ')
# client.create_database('raspi')
print(client.get_list_database())
client.switch_database('raspi')

json = [
    {
        "measurement": "brushEvents",
        "tags": {
            "user": "Carol",
            "brushId": "6c89f539-71c6-490d-a28d-6c5d84c0ee2f"
        },
        "time": "2018-03-28T8:01:00Z",
        "fields": {
            "duration": 127
        }
    },
    {
        "measurement": "brushEvents",
        "tags": {
            "user": "Carol",
            "brushId": "6c89f539-71c6-490d-a28d-6c5d84c0ee2f"
        },
        "time": "2018-03-29T8:04:00Z",
        "fields": {
            "duration": 132
        }
    }
]

if client.write_points(json, time_precision="s") != True:
    print ("ERROR: Write to InfluxDB not successful")
else: 
    print ("data sent")
Retrieved from "https://entorb.net//wiki/index.php?title=InfluxDB&oldid=4900"