Frage : Migrate MySQL nach Influxdb

Dutchman

@ehome sagte in Frage : Migrate MySQL nach Influxdb:

@Dutchman
könntets du vielleicht den export/import etwas genauer erklären ?

influxdb kan man einen import machen mit CSV, die Schwierigkeit : SQL arbeitet mit Tabellen, InfluxDB mit measurements.

Du musst also aus den SQL Tabellen die werte extrahieren, in measurements separieren, die richtigen Datensätze zusammenbringen (dabei den timestamp umrechnen) und dan importieren in influxdb.

@arteck hattest du nicht etwas dafür oder letztendlich auch gelassen ?

Dutch

arteck

@Dutchman ne ich habs gelassen... war mir nicht wert die Zeit da zu inverstieren..

ehome

@Dutchman
ah, ok - jetzt habe ich es verstanden.
Hier mal ein Beispiel für die mysql Abfrage:

SELECT 
	(ts_number.ts*1000000)       as 'time',
    ts_number.ack    as 'ack',
    datapoints.name as 'from',
    ts_number.q        as 'q',
    ts_number.val     as 'val'
from ts_number 
left join datapoints
on ts_number.id=datapoints.id
where datapoints.name="wrapper.0.Temperatur.temp_ab" order by ts_number.ts desc limit 10;

gender

Da ich auch vor dem Problem stand, die SQL-Daten in Influxdb zu importieren, möchte ich hier einmal meinen Workarround darstellen (da ich doch einen ganzen Nachmittag rumprobieren musste, bis es endlich geklappt hat).

1. SQL Daten für CSV-Datei-Export abfragen:

SELECT 
(ts_string.ts*1000000)       as 'time',
ts_string.ack    as 'ack',
datapoints.name as 'from',
ts_string.q        as 'q',
ts_string.val     as 'value'
from ts_string 
left join datapoints
on ts_string.id=datapoints.id
where datapoints.name="mqtt.0.Stromzaehler_Gesamt" AND q=0

Danke an @ehome

2. Ergebnis als "Delimted Text" und mit Feldbegrenzer , exportieren (habe ich mit HeidiSQL gemacht, geht sicher auch mit anderen Programmen)

3. Mithilfe von Python bzw. dem Script "csv_to_line.py" die CSV-Datei nach InfluxDB-CommandLine-Vorgabe in eine txt-Datei importieren.
Also python csv_to_line.py
(kann dauern: Ich habe 2 Millionen Einträge gewandelt - hat ca. 10 Minuten gedauert).

Hier das Python-Script: csv_to_line.py
(hierfür muss pandas in Python installiert sein!!!). In der Datei muss der "Measurement"-Bezeichner sowie der Name eurer CSV-Datei angepasst werden.
Außerdem habe ich das "ack"-Value fest auf true gesetzt. In der SQL-Datenbank ist das eigentlich 1. Problem ist jedoch, dass Influx diese Spalte dann beim Importieren nicht als Boolean ansieht... der Influx-Adapter in Iobroker dieses Field aber nicht als 1 sondern als True einträgt und man sonst keine weiteren Daten über IoBroker in die Datenbank schreiben kann. Das wäre natürlich ungünstig.

4. Unbedingt darauf achten, dass die Textdatei im Unix (LF) - Format ist (ggf. mit bspw. NotePad ändern)

5. In den Kopf der Txt-Datei

# DDL
CREATE DATABASE NameDatabase

# DML
# CONTEXT-DATABASE: NameDatabase

schreiben.
Der obere Teil kann weggelassen werden, falls die Daten in eine vorhandene Datenbank importiert werden sollen.

6. In der Console mit dem Befehl

influx -import -path=import.txt

die Import-Datei importieren.

Wenn alles klappt, zeigt er alle 100000 Einträge eine Erfolgsmeldung an.

Ich weiß nicht, ob es auch einfacher geht (da die anderen das hier nur knapp beschrieben haben), aber so geht es in jedem Fall, wenn auch viele Schritte nötig sind.

schittl

@gender Danke für diese näheren Info's.... Wenn ich deinen Select ausführe über alle Datenpunkte komme ich so auf ca. 800000 Datensätze, davon werden aber nach Aufbereitung nur 400000 importiert... Sollte man das staffeln oder an was kann das liegen? Umlaute o.ä?

2019/12/31 00:29:10 Processed 0 commands
2019/12/31 00:29:10 Processed 425000 inserts
2019/12/31 00:29:10 Failed 428620 inserts
ERROR: 428620 points were not inserted

Bei 1000000 Datensätze werden immer nur wie folgt importiert. Gibt es einen Parameter um die Batchsize zu umgehen oder welcher anderer Ansatz würde funktionieren? Telegraf?

2019/12/31 15:26:46 Processed 1000000 lines.  Time elapsed: 10.569888124s.  Points per second (PPS): 94608
2019/12/31 15:26:46 Processed 0 commands
2019/12/31 15:26:46 Processed 360000 inserts
2019/12/31 15:26:46 Failed 640000 inserts

gender

Also ich habe wie beschrieben ca. 2 Millionen Datensätze in einem Rutsch importieren können. Da gab es keine Fehlermeldung... und alle Hunderttausend Datensätze immer eine Konsolenausgabe. Hat ca. 2 1/2 Minuten gedauert.
Ob es an Umlauten liegt, kann ich dir nicht sagen. Ich hatte auf jeden Fall keine in meinen Datensätzen drin.

Kriegst du in der Konsole keine Fehlermeldung für einzelne Datensätze? Das war bei mir jedenfalls so, sodass ich genauer gucken konnte, wenn er etwas nicht importiert hat.

schittl

@gender In der ini von influx bzw. von der Doku dazu hatte ich nur was gelesen das man die Files splitten sollte bei 10000. Vielleicht liegt es auch an der neuen Version 1.7

siehe hier

„If your data file has more than 5,000 points, it may be necessary to split that file into several files in order to write your data in batches to InfluxDB. We recommend writing points in batches of 5,000 to 10,000 points. Smaller batches, and more HTTP requests, will result in sub-optimal performance. By default, the HTTP request times out after five seconds. InfluxDB will still attempt to write the points after that time out but there will be no confirmation that they were successfully written.“

Fehlermeldung ist folgende:

2020/01/01 17:00:44 error writing batch:  {"error":"partial write: points beyond retention policy dropped=5000"}

gender

Ja, das hatte ich in der Doku auch mal gelesen.... und mich dann einfach gefreut, dass es auch so ging (zumal das Hochladen von Millionen Daten in 10000er-Schritten dann doch ne Zumutung wäre.

Deine Fehlermeldung sagt mir leider nichts und ist mir auch unbekannt (mal abgesehen davon, dass ich auch noch nicht verstanden habe, was es mit dieser retention policy bei influx auf sich hat, die man ja irgendwie optional angeben/einstellen kann, soweit ich das verstanden habe.

Ich habe die Datenbank auf jeden Fall einfach von Iobroker beim erstem Sammeln von Daten erstellen lassen.

schittl

@gender Influx DB wurde bei mir auch beim ersten Sammeln automatisch erstellt. Kann mir auch nicht vorstellen, das man es einzeln machen muss. Komisch ist ja das er immer bei 360000 aufhört. Ist bestimmt irgendeine neue Grenze. Welche Version nutzt Du bzw. in welcher Version klappte dein Import? Vielleicht kennt jemand noch einen anderen Ansatz z.B Telegraf? Wäre schon Schade, wenn ich die Daten von ca. 2 Jahren verlieren würde (bittl mehr als 8 GB)

gender

Ich habe Version 1.7.9, installiert im Docker.
Habe das auch erst vor ein paar Tagen alles installiert bzw. umgestellt auf Influxdb, daher sollte die Version aktuell sein.

Ich hatte auf jeden Fall Probleme mit den "ack"-Werten, da die in meinen MariaDB-Tabellen aus 0en oder 1en bestehen und Influx daraus dann einen Zahlenwert und keinen Boolean gemacht hat.
Daher wird oben im Python-Skript einfach alles auf true gesetzt (und die Zeilen mit 0er-Werten hatte ich ausgeschlossen).

schittl

@gender Bei mir auch die neueste nur in einem Container unter Proxmox... Dann verstehe ich das ja gar nicht mehr... Trotzdem danke und vielleicht hat jemand ja noch eine andere Idee bzw. kann mal kurz aufzeigen wie er die Migration großer DB's vorgenommen hat...

Könnte mir ja auch vorstellen ne Batchverarbeitung zu schreiben nur fehlt mir dazu erst der Ansatz/Knowhow... Idee wäre jeden Datenpunkt einzeln in eine Datei zu schreiben (geht sowas mit dem Javascript Adapter getHistory etc.?) und dann einzeln per Bash etc. in Influx einzulesen...

schittl

Wollte hier nur mal hinterlassen, dass ich die Daten (über 90% ca. 4 GB) migrieren konnte mit den Beispiel, was weiter oben von @gender genannt wurde. Der einzige Fehler den man nicht machen darf ist, dass Du erst die Daten migrieren musst ehe du in ioBroker den Adapter aktivierst / einrichtest. Ansonsten ging das sehr gut:

Alle Daten in mehreren csv-Dateien aus mysql exportiert (so ca. 2000000 pro Datei)
Per python csv_to_line.py alle Dateien konvertiert in txt (wichtig dabei # DML und # CONTEXT-DATABASE: NameDatabase ergänzen in den ersten 2 Zeilen)
Per influx-Befehl die konvertierten Dateien eingelesen
ioBroker-Adapter aktivieren

UlliJ

Moin zusammen,

mein erster post in diesem Forum...
Zu aller erst vielen Dank für diese tolle Plattform und den vielen die mit enormem Aufwand das smarte home nach vorne treiben, tolle Ideen weiter geben, absolut geile Visualisierungen bauen usw. Ich habe bislang selten ein Forum gesehen, in dem so konstruktiv und freundlich miteinander umgegangen wird...weiter so

Seit ca 2 Jahren betreibe ich Iobroker und habe bislang die Daten zuerst in MySQL und später dann in PostgreSQL abgelegt. Über die Zeit wurden die ganzen Abfragen und Diagramme mit Grafana nach und nach deutlich träger. Nach kurzer Testphase mit InfluxDB war die Frage geklärt
Mit dem Wechsel auf InfluxDB stellt sich die Frage für den nicht Programmierer wie die Daten von alt nach neu...

Die genannten Lösungen haben bei mir nicht zuverlässig funktioniert also die nächste Klappe

In einem SQL GUI (phpmyadmin, HeidiSQL oder DBeaver, etc) oder auf der SQL-Konsole die zu exportierenden Daten in einem View zur Verfügung stellen
Der View enthält alles was Influx benötigt

create view xxx as
SELECT datapoints.name,
	ts_number.ack    as "ack",
    (ts_number.q*1.0)       as "q",
	sources.name as "from",
    (ts_number.val*1.0)     as "value",
    (ts_number.ts*1000000) as "time"
from ts_number 
left join datapoints
on ts_number.id=datapoints.id
left join sources
on ts_number._from=sources.id
where datapoints.id>=70 and datapoints.id<=73 and q=0
order by ts_number.ts desc;

"xxx" Name des View anpassen
in der WHERE Klausel die Datenpunkte entsprechend anpassen, einen Bereich wie hier oder auch eine einzelne id
Für kleine Datenbestände kann die WHERE Klausel auch auf "WHERE q=0" gesetzt werden und über den View in einem Rutsch in Influx übertragen werden.
Beispiel:

Python installieren und ggf. die Zusätze installieren

psycopg2 für PostgreSQL bzw. MySQLdb für MySQL
influxdb-client und influxdb.

das Skript von Muntazir Fadhel "Easily Migrate Postgres/MySQL Records to InfluxDB" nehmen (modifizierte Variante hier)

### MySQL DB info ###
#import MySQLdb
#conn = MySQLdb.connect(host="localhost",  # your host, usually localhost
#                     user="john",         # your username
#                     passwd="megajonhy",  # your password
#                     db="jonhydb")        # name of the data base


### PostgreSQL DB info ###
import psycopg2
import psycopg2.extras

#####
# connection data for PostgreSQL
conn = psycopg2.connect("dbname=xxx user=xxx password=xxx host=xxx.xxx.xxx.xxx port =5432")
#####

# InfluxDB info #
from influxdb import InfluxDBClient
#
#####connection data for InfluxDB#####
influxClient = InfluxDBClient(host='xxx.xxx.xxx.xxx', port=8086, username='xxx', password='xxx', database='xxx')
#####
#influxClient.delete_database(influx_db_name)
#influxClient.create_database(influx_db_name)

# dictates how columns will be mapped to key/fields in InfluxDB
schema = {
    "time_column": "time", # the column that will be used as the time stamp in influx
    "columns_to_fields" : ["ack","q", "from","value"], # columns that will map to fields 
    # "columns_to_tags" : ["",...], # columns that will map to tags
    "table_name_to_measurement" : "name", # table name that will be mapped to measurement
    }

'''
Generates an collection of influxdb points from the given SQL records
'''
def generate_influx_points(records):
    influx_points = []
    for record in records:
        #tags = {}, 
        fields = {}
        #for tag_label in schema['columns_to_tags']:
        #   tags[tag_label] = record[tag_label]
        for field_label in schema['columns_to_fields']:
            fields[field_label] = record[field_label]
        influx_points.append({
            "measurement": record[schema['table_name_to_measurement']],
            #"tags": tags,
            "time": record[schema['time_column']],
            "fields": fields
        })
    return influx_points



# query relational DB for all records
curr = conn.cursor('cursor', cursor_factory=psycopg2.extras.RealDictCursor)
# curr = conn.cursor(dictionary=True)
#####
# SQL query for PostgreSQL, syntax for MySQL differs
# query provide desired columns as a view on the sql server

# request data from SQL, adjust ...from <view name>
curr.execute("Select * from xxx;")
#####
row_count = 0
# process 1000 records at a time
while True:
    print("Processing row #" + str(row_count + 1))
    selected_rows = curr.fetchmany(1000)
    influxClient.write_points(generate_influx_points(selected_rows))
    row_count += 1000
    if len(selected_rows) < 1000:
        break
conn.close()

Im Skript sind

in Zeile 15 die Datenbankverbindung für Postgre anzupassen,
in Zeile 22 die Datenbankverbindung für Influx
weiter unten Zeile 65 statt xxx ist der Name des Views einzutragen der auf der Datenbank mit der Abfrage aus 1 erzeugt wurde
curr.execute("Select * from xxx;")
der Abschnitt "schema" gibt das mapping der Spalten vor und muss nur angepasst werden wenn die Spaltenbezeichner in der Abfrage aus 1 geändert werden
für die MySQL user sind die Verbindungsdaten in den Zeilen 2-6 auskommentiert und an eure Umgebung anzupassen.

In der Python Umgebung das Skript starten...alle 1000 Zeilen/Datensätze kommt eine Fortschrittsmeldung. Während die Abfrage auf der SQL Seite läuft steht da "Processing row # 1"

Bei umfangreichen Abfragen und je nach Umgebung kann das Ganze ein wenig dauern. Teilweise dauerten bei mir die SQL Abfragen bis zu 20min bei 20Mio Datensätzen (Nuc i3 mit 12GB für die SQL VM unter proxmox). Das Skript läuft zuverlässig und mehrere 10 Mio Datensätze von A nach B wurden migriert. Vielleicht kann es jemand brauchen und nochmals Danke für das tolle Projekt und die hilfsbereiten Menschen hier
VG
Ulli

arteck

@UlliJ DANKÖÖÖÖÖÖÖÖÖÖÖÖÖÖÖÖ ich war zu faul zu

UlliJ

@arteck Gerne doch habe auch mehr Probleme auf der Skriptseite, die Datenbank geht eigentlich ganz gut...bis man mit Influx anfängt

JackGruber

@UlliJ danke für die Idee mit dem Script.
Hab das script etwas angepasst, damit alle Daten übernommen werden und diese auch auf einem leistungsschwachen Rasperry Pi ausgelesen werden können.
Hab so ca. 22 Millionen Datensätze in die InfluxDB gepackt.

GitHub iobroker_mysql_2_influxdb

UlliJ

@JackGruber danke für die coole Ergänzung, habe ich gerade erst gesehen. Das hätte meine Möglichkeiten schon wieder weit überfordert

MezzoDO

Ich habe nun versucht mit dem Git-Projekt von @JackGruber meine Daten zu Influx zu migrieren.

Nach anfänglichen Schwierigkeiten scheint es augenscheinlich durch zu laufen.

Tatsächlich werden die Daten jedoch nicht migriert und ich kenne mich mit Python nicht aus, um mir z. B. Debug-Points o.ä. zu setzen...

Was kann ich tun, um zu sehen, wo es denn nun hakt?

JackGruber

@MezzoDO was ist genau das problem?
Gibt es ausgaben oder fehlermeldungen?

MezzoDO

@JackGruber
ich habe es mittlerweile geschafft, die Daten zu importieren.
Allerdings besteht nun, wie oben schon beschrieben das Problem mit den Bool-Werten.
Wenn ich einerseits eine leere Datenbank mit meinem Import befülle ist "ack" ein Integer.
Lasse ich die InfluxDB erst durch ioBroker anlegen ist es korrekterweise ein bool, aber dann bekomme ich die Fehlermeldung:

influxdb.exceptions.InfluxDBClientError: 400: {"error":"partial write: field type conflict: input field \"ack\" on measurement \"info.0.sysinfo.cpu.currentLoad.currentload\" is type integer, already exists as type boolean dropped=1000"}

Ich glaube also dass ich im besten Falle den Datentypen beim Import angeben sollte, oder?
Nur: wie?

NEWS

Frage : Migrate MySQL nach Influxdb

Support us

383

32.4k

81.4k

1.3m