Erstellen einer Zeitreihendatenbank mit TDEngine und GraphQL

Motivation und Einführung

Als Teil des Softwareteams von Nevados bauen wir eine Betriebs- und Überwachungsplattform für den Nevados All Terrain Tracker®. Ein Solartracker ist ein Gerät, das ein Solarpanel auf die Sonne ausrichtet. Jeder Solartracker sendet ständig Statusinformationen und Messwerte wie den aktuellen Winkel, die Temperatur, die Spannungen usw. an unsere Plattform und wir müssen diese Informationen zur Analyse und Visualisierung speichern. Wenn der Tracker so konfiguriert ist, dass er alle 5 Sekunden Daten sendet, haben wir 17.280 Datenpunkte pro Tracker und Tag, 518.400 Datenpunkte pro Tracker und Monat. Das fasst viele Informationen zusammen. Diese Art von Daten nennt man „Zeitreihendaten“ und wie für alle komplexen Probleme in der Software gibt es dafür mehrere Lösungen (Zeitreihendatenbanken). Die bekanntesten sind InfluxDB und TimescaleDB. Für unsere Plattform haben wir uns für die Arbeit mit TDEngine entschieden, einem relativ neuen Produkt, das für IoT-Anwendungen optimiert ist und mit der SQL-Abfragesprache arbeitet.

Für diese Entscheidung gab es mehrere Argumente: TDEngine

• ist Open Source
• ist für IoT-Anwendungen optimiert
• verwendet SQL, eine Sprache, mit der wir vertraut sind
• ist als Managed Service verfügbar und wir können uns auf die Entwicklung unserer Anwendung konzentrieren
• lässt sich einfach lokal über Docker ausführen

In diesem Artikel gehen wir durch die Einrichtung einer TDEngine-Datenbank und -Tabellen und wie man ein GraphQL-Schema erstellt, das es uns ermöglicht, die Daten von verschiedenen Clients und Anwendungen abzufragen.

Erste Schritte mit TDEngine

Der einfachste Weg, mit TDEngine zu beginnen, ist die Nutzung ihres Cloud-Dienstes. Gehen Sie zur TDEngine und erstellen Sie ein Konto. Sie verfügen über einige öffentliche Datenbanken, die wir nutzen können, was sich hervorragend zum Erstellen einer Demo oder zum Experimentieren mit Abfragen eignet.

Wenn Sie TDEngine lokal ausführen möchten, können Sie mit dem Docker-Image und Telegraf Daten aus verschiedenen Quellen abrufen und an die Datenbank senden, z. B. Systeminformationen, Ping-Statistiken usw.

 version: '3.9' services: tdengine: restart: always image: tdengine/tdengine:latest hostname: tdengine container_name: tdengine ports: - 6030:6030 - 6041:6041 - 6043-6049:6043-6049 - 6043-6049:6043-6049/udp volumes: - data:/var/lib/taos telegraf: image: telegraf:latest links: - tdengine env_file: .env volumes: - ./telegraf.conf:/etc/telegraf/telegraf.conf

Sehen Sie sich die offizielle Dokumentation zur Telegraf-Konfiguration und die TDEngine-Dokumentation zu Telegraf an. Kurz gesagt würde dies etwa so aussehen, um eine Verbindung zu einem MQTT-Thema herzustellen:

 [agent] interval = "5s" round_interval = true omit_hostname = true [[processors.printer]] [[outputs.http]] url = "http://127.0.0.1:6041/influxdb/v1/write?db=telegraf" method = "POST" timeout = "5s" username = "root" password = "taosdata" data_format = "influx" [[inputs.mqtt_consumer]] topics = [ "devices/+/trackers", ]

Anstatt alles lokal einzurichten und darauf zu warten, dass die Datenbank mit Informationen gefüllt wird, verwenden wir für diesen Artikel die öffentliche Datenbank, die Schiffsbewegungen von den fünf großen US-Häfen enthält.

Verwendung von TDEngine mit öffentlichen Schiffsbewegungsdaten

Standardmäßig verfügen die Tabellen in TDEngine über ein implizites Schema, was bedeutet, dass sich das Schema an die Daten anpasst, die in die Datenbank geschrieben werden. Dies eignet sich hervorragend zum Bootstrapping, aber irgendwann möchten wir zu einem expliziten Schema wechseln, um Probleme mit eingehenden Daten zu vermeiden. Etwas gewöhnungsbedürftig ist das Konzept der Super Tables (kurz „STable“). In TDEngine gibt es Tags (Schlüssel) und Spalten (Daten). Für jede Tastenkombination wird eine „Tabelle“ erstellt. Alle Tabellen sind in der STable gruppiert.

Wenn man sich die vessel ansieht, gibt es eine STable namens ais_data , die viele Tabellen enthält. Normalerweise möchten wir keine Abfragen pro Tabelle durchführen, sondern immer STable verwenden, um akkumulierte Daten aus allen Tabellen abzurufen.

TDEngine verfügt über eine Funktion DESCRIBE , mit der wir das Schema einer Tabelle oder STable überprüfen können. Die ais_data haben das folgende Schema:

Die STable verfügt über zwei Schlüssel und sechs Datenspalten. Die Schlüssel sind die mmsi und der name . Wir können reguläre SQL-Anweisungen verwenden, um die Daten abzufragen:

 SELECT ts, name, latitude, longitude FROM vessel.ais_data LIMIT 100; ts name latitude longitude 2023-08-11T22:07:02.419Z GERONIMO 37.921673 -122.40928 2023-08-11T22:21:48.985Z GERONIMO 37.921688 -122.40926 2023-08-11T22:25:08.784Z GERONIMO 37.92169 -122.40926 ...

Bedenken Sie, dass Zeitreihendaten normalerweise sehr umfangreich sind, daher sollten wir die Ergebnismenge immer begrenzen. Es gibt einige zeitreihenspezifische Funktionen, die wir verwenden können, z. B. PARTITION BY , die Ergebnisse nach Schlüssel gruppiert und nützlich ist, um die neuesten Aktualisierungen einzelner Schlüssel zu erhalten. Zum Beispiel:

 SELECT last_row(ts, name, latitude, longitude) FROM vessel.ais_data PARTITION BY name; ts name latitude longitude 2023-09-08T13:09:34.951Z SAN SABA 29.375961 -94.86894 2023-09-07T18:05:01.230Z SELENA 33.678585 -118.1954 2023-09-01T17:23:24.145Z SOME TUESDAY 33.676563 -118.230606 ... 

Für weitere Beispiele empfehle ich die Lektüre der SQL-Dokumentation . Bevor wir fortfahren, gehen Sie zu „Programmierung“, „Node.js“ und rufen Sie Ihre Variablen TDENGINE_CLOUD_URL und TDENGINE_CLOUD_TOKEN ab.

GraphQL mit Nexus.js, Fastify und Mercurius

GraphQL ist heutzutage ziemlich bekannt und es gibt viele gute Artikel darüber. Wir haben uns für die Technologie entschieden, da wir Informationen aus verschiedenen Quellen sammeln und verarbeiten und GraphQL es uns ermöglicht, diese transparent in einer einzigen API zu kombinieren.

Wir verwenden das erstaunliche Fastify- Framework (mittlerweile die Standardauswahl für Node.js-Anwendungen) und den Mercurius- Adapter. Die Teams von Mercurius und Fastify haben für ein nahtloses Erlebnis zusammengearbeitet und es ist eine großartige Wahl für GraphQL-APIs mit Fokus auf Leistung. GraphQL Nexus ist ein Tool zum Erstellen/Generieren des Schemas und der Resolver, sodass wir nicht alles von Hand schreiben müssen.

Es muss noch ein bisschen Setup-Code usw. erledigt werden, den ich hier überspringe. Ein vollständiges Beispiel finden Sie auf GitHub – tdengine-graphql-example .

Ich möchte in diesem Artikel auf zwei Dinge näher eingehen, die ziemlich spezifisch sind:

1. die TDEngine-Abfragebibliothek
2. das GraphQL-Schema mit Nexus

TDEngine-Abfragebibliothek

TDEngine verfügt über eine Node.js-Bibliothek , die es uns ermöglicht, die Datenbank abzufragen. Dadurch ist es einfach, eine Verbindung herzustellen und Anfragen zu senden. Leider ist es etwas schwierig, mit den Antworten zu arbeiten. Also haben wir einen kleinen Wrapper geschrieben:

 'use strict' import tdengine from '@tdengine/rest' import { tdEngineToken, tdEngineUrl } from '../config.js' import parseFields from 'graphql-parse-fields' const { options: tdOptions, connect: tdConnect } = tdengine tdOptions.query = { token: tdEngineToken } tdOptions.url = tdEngineUrl export default function TdEngine(log) { this.log = log const conn = tdConnect(tdOptions) this.cursor = conn.cursor() } TdEngine.prototype.fetchData = async function fetchData(sql) { this.log.debug('fetchData()') this.log.debug(sql) const result = await this.cursor.query(sql) const data = result.getData() const errorCode = result.getErrCode() const columns = result.getMeta() if (errorCode !== 0) { this.log.error(`fetchData() error: ${result.getErrStr()}`) throw new Error(result.getErrStr()) } return data.map((r) => { const res = {} r.forEach((c, idx) => { const columnName = columns[idx].columnName .replace(/`/g, '') .replace('last_row(', '') .replace(')', '') if (c !== null) { res[columnName] = c } }) return res }) }

Dies gibt ein TDEngine-Objekt zurück, das an den GraphQL-Kontext übergeben werden kann. Wir verwenden hauptsächlich die Funktion fetchData , mit der wir eine SQL-Abfrage übergeben und die Ergebnisse als Array von Objekten zurückerhalten können. TDEngine gibt die Metadaten (Spalten), Fehler und Daten separat zurück. Wir verwenden die Metadaten, um die Spalten einer regulären Liste von Objekten zuzuordnen. Ein Sonderfall ist hier die Funktion last_row . Die Spalten werden als last_row(ts) , last_row(name) usw. zurückgegeben und wir möchten den last_row Teil entfernen, damit das Attribut 1:1 dem GraphQL-Schema zugeordnet wird. Dies geschieht im Teil columnName.replace .

GraphQL-Schema

Leider gibt es keinen Schemagenerator wie Postgraphile für TDEngine und wir möchten kein reines GraphQL-Schema schreiben und verwalten, daher verwenden wir Nexus.js, um uns dabei zu helfen. Wir beginnen mit zwei Grundtypen: VesselMovement und Timestamp (ein Skalartyp). Timestamp und TDDate sind zwei verschiedene Typen, um das Datum als Zeitstempel oder als Datumszeichenfolge anzuzeigen. Dies ist für die Clientanwendung (und während der Entwicklung) nützlich, da sie entscheiden kann, welches Format verwendet werden soll. asNexusMethod können wir den Typ als Funktion im VesselMovement Schema verwenden. Wir können das TDDate direkt hier in der Typdefinition auflösen, um den ursprünglichen ts Zeitstempelwert zu verwenden.

 import { scalarType, objectType } from 'nexus' export const Timestamp = scalarType({ name: 'Timestamp', asNexusMethod: 'ts', description: 'TDEngine Timestamp', serialize(value) { return new Date(value).getTime() } }) export const TDDate = scalarType({ name: 'TDDate', asNexusMethod: 'tdDate', description: 'TDEngine Timestamp as Date', serialize(value) { return new Date(value).toJSON() } }) export const VesselMovement = objectType({ name: 'VesselMovement', definition(t) { t.ts('ts') t.tdDate('date', { resolve: (root) => root.ts }) t.string('mmsi') t.string('name') t.float('latitude') t.float('longitude') t.float('speed') t.float('heading') t.int('nav_status') } })

Für Zeitreihentypen verwenden wir das Suffix Movement oder Series für eine klare Trennung von relationalen und Zeitreihentypen in der Schnittstelle.

Jetzt können wir die Abfrage definieren. Wir beginnen mit einer einfachen Abfrage, um die neuesten Bewegungen von TDEngine abzurufen:

 import { objectType } from 'nexus' export const GenericQueries = objectType({ name: 'Query', definition(t) { t.list.field('latestMovements', { type: 'VesselMovement', resolve: async (root, args, { tdEngine }, info) => { const fields = filterFields(info) return tdEngine.fetchData( `select last_row(${fields}) from vessel.ais_data partition by mmsi;` ) } }) } }) 

GraphiQL ist ein großartiges Tool zum Testen der API und zum Erkunden des Schemas. Sie können es aktivieren, indem Sie graphiql.enabled = true in Mercurius übergeben. Mit der Abfrage können wir die neuesten Schiffsbewegungen gruppiert nach mmsi sehen. Gehen wir aber noch etwas weiter. Einer der größten Vorteile von GraphQL besteht darin, dass es eine transparente Ebene für den Client oder die Anwendung darstellt. Wir können Daten aus mehreren Quellen abrufen und sie in demselben Schema kombinieren.

Leider konnte ich keine einfache/kostenlose API mit umfangreichen Schiffsinformationen finden. Es gibt Sinay , aber sie geben in ihrer Vessel-Antwort (die wir bereits in TDEngine haben) nur den name , mmsi und imo an. Für das Beispiel gehen wir davon aus, dass wir den name nicht in unserer Datenbank haben und ihn von Sinay abrufen müssen. Mit dem imo könnten wir auch CO2-Emissionen für ein Schiff abfragen oder eine andere API verwenden, um ein Bild, die Flagge oder andere Informationen abzurufen, die alle im Vessel kombiniert werden können.

 export const Vessel = objectType({ name: 'Vessel', definition(t) { t.string('mmsi') t.string('name') t.nullable.string('imo') t.list.field('movements', { type: 'VesselMovement' }) } })

Wie Sie hier sehen können, können wir den Zeitreihendaten von TDEngine eine Liste mit movements hinzufügen. Wir fügen eine weitere Abfrage hinzu, um die Schiffsinformationen abzurufen, und der Resolver ermöglicht es uns, die Daten von TDEngine und Sinay zu kombinieren:

 t.field('vessel', { type: 'Vessel', args: { mmsi: 'String' }, resolve: async (root, args, { tdEngine }, info) => { const waiting = [ getVesselInformation(args.mmsi), tdEngine.fetchData( `select * from vessel.ais_data where mmsi = '${args.mmsi}' order by ts desc limit 10;` ) ] const results = await Promise.all(waiting) return { ...results[0][0], movements: results[1] } } }) 

🎉 und hier haben wir eine funktionierende GraphQL-API, die Zeilen von TDEngine für ein bestimmtes von uns angefordertes Schiff zurückgibt. getVesselInformation() ist ein einfacher Wrapper zum Abrufen von Daten von Sinay. Wir fügen die TDEngine-Ergebnisse dem movements hinzu und GraphQL kümmert sich um den Rest und ordnet alles dem Schema zu.

Hinweis: SQL-Injection

Wie bei jeder SQL-Datenbank müssen wir bei Benutzereingaben vorsichtig sein. Im obigen Beispiel verwenden wir die mmsi Eingabe direkt, was diese Abfrage anfällig für SQL-Injections macht. Für das Beispiel ignorieren wir dies vorerst, aber in „realen“ Anwendungen sollten wir Benutzereingaben immer bereinigen. Es gibt mehrere kleine Bibliotheken zum Bereinigen von Zeichenfolgen. In den meisten Fällen verlassen wir uns nur auf Zahlen (Paginierung, Limit usw.) und Aufzählungen (Sortierreihenfolge), die GraphQL für uns überprüft.

Vielen Dank an Dmitry Zaets für den Hinweis!

Optimierungen

Es gibt ein paar Dinge, die den Rahmen dieses Artikels sprengen, die ich aber kurz erwähnen möchte:

Pothos als spiritueller Nachfolger von Nexus.js

Als wir mit dem Projekt begannen, war Nexus.js die beste Wahl, um unser GraphQL-Schema zu generieren. Obwohl stabil und einigermaßen funktionsreich , mangelt es an Wartung und Updates. Es gibt einen Plugin-basierten GraphQL-Schema-Builder namens Pothos , der etwas moderner ist und aktiv gepflegt wird. Wenn Sie ein neues Projekt starten, empfehle ich wahrscheinlich die Verwendung von Pothos anstelle von Nexus.js.

Vielen Dank an Mo Sattler für den Hinweis!

Feldlöser

Wie Sie oben im Vessel sehen können, werden beide Datenquellen sofort abgerufen und verarbeitet. Das heißt, wenn die Abfrage nur den name betrifft, rufen wir trotzdem die movements für die Antwort ab. Und wenn es sich bei der Anfrage nur um die movements handelt, holen wir uns trotzdem den Namen von Sinay und zahlen möglicherweise für die Anfrage.

Das ist ein GraphQL-Anti-Pattern und wir können die Leistung verbessern, indem wir die Feldinformationen verwenden, um nur die angeforderten Daten abzurufen. Resolver haben die Feldinformationen als viertes Argument, aber es ist ziemlich schwierig, mit ihnen zu arbeiten. Stattdessen können wir graphql-parse-fields verwenden, um ein einfaches Objekt der angeforderten Felder abzurufen und die Resolver-Logik anzupassen.

SQL-Abfrageoptimierungen

In unseren Beispielabfragen verwenden wir select * um alle Spalten aus der Datenbank abzurufen, auch wenn sie nicht benötigt werden. Das ist offensichtlich ziemlich schlecht und wir können denselben Feldparser verwenden, um die SQL-Abfragen zu optimieren:

 export function filterFields(info, context) { const invalidFields = ['__typename', 'date'] const parsedFields = parseFields(info) const fields = context ? parsedFields[context] : parsedFields const filteredFields = Object.keys(fields).filter( (f) => !invalidFields.includes(f) ) return filteredFields.join(',') }

Diese Funktion gibt eine durch Kommas getrennte Liste von Feldern aus den GraphQL-Informationen zurück.

 const fields = filterFields(info) return tdEngine.fetchData( `select last_row(${fields}) from vessel.ais_data partition by mmsi;` )

Wenn wir ts , latitude und longitude anfordern, würde die Abfrage so aussehen:

 select last_row(ts, latitude, longitude) from vessel.ais_data partition by mmsi;

Bei nur wenigen Spalten in dieser Tabelle spielt dies möglicherweise keine große Rolle, bei mehr Tabellen und komplexeren Abfragen kann dies jedoch einen großen Unterschied in der Anwendungsleistung bewirken.

Zeitreihenfunktionen

TDEngine verfügt über einige zeitreihenspezifische Erweiterungen, die zur Verbesserung der Leistung verwendet werden sollten. Um beispielsweise den neuesten Eintrag abzurufen, reicht eine herkömmliche SQL-Abfrage aus:

 SELECT ts, name, latitude, longitude FROM vessel.ais_data order by ts desc limit 1;

Die Ausführung dauert 653 ms, während die „TDEngine“-Abfrage nur 145 ms benötigt:

 SELECT last_row(ts, name, latitude, longitude) FROM vessel.ais_data;

Für jede Tabelle gibt es Konfigurationsoptionen zur Optimierung für last_row/first_row-Funktionen und andere Cache-Einstellungen. Ich empfehle die Lektüre der TDEngine-Dokumentation .

Abschluss

Die einfache Version: In diesem Artikel haben wir eine TDEngine-Zeitreihendatenbank eingerichtet und ein GraphQL-Schema definiert, damit Clientanwendungen Daten verbinden und abfragen können.

Es steckt noch viel mehr dahinter. Wir haben ein Standardprojekt, um komplexe Zeitreihendaten mit relationalen Daten in einer transparenten Schnittstelle zu kombinieren. Bei Nevados verwenden wir PostgreSQL als Primärdatenbank und rufen Zeitreihendaten auf die gleiche Weise ab wie im movement oben. Dies ist eine großartige Möglichkeit, Daten aus mehreren Quellen in einer einzigen API zu kombinieren. Ein weiterer Vorteil besteht darin, dass die Daten nur auf Anfrage abgerufen werden, was der Clientanwendung viel Flexibilität verleiht. Zu guter Letzt fungiert das GraphQL-Schema als Dokumentation und Vertrag, sodass wir das Kästchen „API-Dokumentation“ problemlos ankreuzen können.

Wenn Sie Fragen oder Kommentare haben , wenden Sie sich bitte an BlueSky oder beteiligen Sie sich an der Diskussion auf GitHub .