In dieser Serie zeigen wir am Beispiel von Leistungsdaten einer PV-Anlage, die in Form von CSV-Dateien vorliegen, wie diese in MongoDB Atlas importiert, aufbereitet und in Charts dargestellt werden können. Dieser Artikel beschreibt den ersten Schritt, den Import der Daten ins Cluster und was dabei zu beachten ist.

CSV-Format - oder CSV-Formate?

Das CSV-Format ist ein Datenformat, das häufig eingesetzt wird, um Daten zwischen verschiedenen Systemen auszutauschen. Auch in unserem Fall liefert der Wechselrichter - neben einer Cloud-Anwendung - detaillierte Daten in Form von CSV-Dateien. Jeder, der bereits unterschiedliche CSV-Dateien in Excel importiert hat, weiß, dass CSV nicht immer CSV heißt, sondern dass der Aufbau von CSV-Dateien selten einheitlich ist. Beispielweise werden unterschiedliche Trennzeichen (Komma, Semikolon) verwendet und Texte mitunter durch Anführungszeichen gekennzeichnet.

In einer Datenbank wollen wir natürlich mit gut passenden Datentypen arbeiten; dies betrifft vor allem Datumswerte und Zahlen. Weil das CSV-Format an dieser Stelle wenig Unterstützung bietet, ist u.U. ein Nachbearbeitungsschritt notwendig.

CSV in MongoDB importieren

Es gibt zwei Möglichkeiten, mit denen CSV-Dateien in eine MongoDB-Collection importiert werden können:

• Die GUI MongoDB Compass
• Das Tool mongoimport

Welche Möglichkeit besser geeignet ist, hängt einerseits davon ab, wie häufig der Import erfolgen soll und andererseits auch von der Struktur der CSV-Datei. Zum Prototyping bietet sich auf jeden Fall die Verwendung von MongoDB Compass als einfache Möglichkeit an, um erst einmal ein Gefühl für die Daten zu erhalten.

Im Gegensatz zu mongoimport unterstützt die GUI die Auswahl des Separators - falls Ihre Datei beispielsweise mit dem Semikolon als Feldtrenner arbeitet, kann man dies in der GUI direkt auswählen - in mongoimport fehlt diese Option, so dass die Datei im Vorhinein entsprechend angepasst werden muss.

mongoimport kann jedoch im Gegensatz zur GUI in einem Skript eingesetzt werden und der Import so automatisiert werden. Ein klarer Vorteil, wenn die Dateien häufig importiert werden sollen. Das Tool bietet darüber hinaus auch erweiterte Funktionen wie beispielsweise Merge oder Upsert der Daten, um doppelte Werte zu vermeiden.

Weitere Informationen zum Import von CSV-Dateien mit MongoDB Compass finden Sie in der Dokumentation. Diese enthält auch detaillierte Informationen zu mongoimport und einen interessanten Artikel, der einige Szenarien bei der Verwendung des Tools beschreibt.

Anpassung des Datenformats

Gerade bei der Verwendung von CSV ist es wahrscheinlich, dass die Struktur der Daten direkt nach dem Import noch nicht ganz optimal ist. In unserem Fall wollten wir das Datum in der Zeitzone anpassen und in der Datenbank als UTC hinterlegen, die elektrische Leistung von Watt in Kilowatt umrechnen und diese in einem Subdokument zusammenfassen. Daher erfolgt der Import zuerst auf relativ einfache Weise flach in eine Collection.

Per Trigger wird stündlich ein kleines Skript ausgeführt, das folgende Aufgaben übernimmt:

• Setzen eines insertedAt-Felds auf den aktuellen Zeitpunkt, so dass die Einträge in der Import-Collection per TTL-Index nach einer konfigurierbaren Zeitspanne entfernt werden. Dies stellt sicher, dass der Speicherverbrauch im Rahmen bleibt, aber hält Daten zumindest ein paar Tage vor, um im Problemfall auf Ursachensuche zu gehen.
• Per Aggregation-Pipeline werden die importierten Rohdaten in eine Collection gemergt, die die relevanten Daten in der aufbereiteten Form dauerhaft speichert. Um das Skript übersichtlich zu halten, haben wir den Großteil der Queue in einer View abgelegt.

Die folgende Übersicht zeigt das Skript:

exports = async function () {
  const pv_import = context.services
    .get("Logger")
    .db("Energy")
    .collection("pv_import");
  const updateResult = await pv_import.updateMany(
    { insertedAt: { $exists: false } },
    { $set: { insertedAt: new Date() } }
  );
  if (updateResult.modifiedCount) {
    const pv_view_ingest = context.services
      .get("Logger")
      .db("Energy")
      .collection("pv_view_ingest");
    await pv_view_ingest
      .aggregate([
        {
          $merge: {
            into: "pv_data",
            on: "_id",
          },
        },
      ])
      .toArray();
  }
};

Das _id-Feld enthält am Ende der Aggregation den Zeitstempel des Eintrags. Durch die $merge-Stage auf Basis dieses Feldes werden Duplikate überschrieben, so dass für jeden Zeitstempel nur ein Eintrag in pv_data vorhanden ist, auch wenn dieser mehrfach in den CSV-Dateien enthalten war.

Nach dem Import sind die Daten anschließend in der Collection pv_data in folgender Struktur abgelegt:

{
  "_id": {
    "$date": "2024-03-18T08:15:00.000Z"
  },
  "power_kW": {
    "dcA": 0.228268,
    "dcB": 0.344431,
    "bat": -0.305667,
    "load": 0.260831,
    "gridFeed": 0.001317,
    "gridLoad": 0.001227
  },
  "soc": 41.82808,
  "socTarget": 97
}

Zusammenfassung

In diesem ersten Schritt haben wir die Daten zuerst aus den CSV-Dateien importiert. Um ein passenderes Datenmodell zu erreichen wurden diese danach per Aggregation in eine neue Collection übernommen. Die Ausgangsdaten in pv_import werden mit einem Import-Zeitstempel versehen, auf dem ein TTL-Index erstellt ist. Dieser stellt sicher, dass die Ausgangsdaten in pv_import nach einigen Tagen gelöscht werden, um Speicherplatz zu sparen.

Die Daten in pv_data sind aktuell noch sehr fein granular, sind für uns aber eine gute Ausgangsposition, um sie auf Stunden- oder Tagesbasis zusammenzufassen. Wie wir dabei vorgehen, zeigen wir im nächsten Artikel der Serie.

---