Creación de una base de datos de series temporales con TDEngine y GraphQL

Motivación e introducción

Como parte del equipo de software de Nevados, estamos construyendo una plataforma de operaciones y monitoreo para Nevados All Terrain Tracker®. Un seguidor solar es un dispositivo que orienta un panel solar hacia el sol. Cada seguidor solar envía constantemente información y lecturas de estado, como el ángulo actual, temperatura, voltajes, etc. a nuestra plataforma y necesitamos almacenar esta información para su análisis y visualización. Si el rastreador está configurado para enviar datos cada 5 segundos, tenemos 17 280 puntos de datos por rastreador por día, 518 400 puntos de datos por rastreador por mes. Eso resume mucha información. Este tipo de datos se denomina "datos de series temporales" y, como para todos los problemas complejos del software, existen varias soluciones (bases de datos de series temporales). Los más famosos son InfluxDB y TimescaleDB. Para nuestra plataforma, decidimos trabajar con TDEngine , un producto relativamente nuevo que está optimizado para aplicaciones de IoT y funciona con el lenguaje de consulta SQL.

Hubo varios argumentos para esta decisión: TDEngine

• es de código abierto
• está optimizado para aplicaciones de IoT
• utiliza SQL, que es un lenguaje con el que estamos familiarizados
• está disponible como un servicio administrado y podemos centrarnos en construir nuestra aplicación
• es fácil de ejecutar localmente a través de Docker

En este artículo, analizaremos la configuración de una base de datos y tablas de TDEngine y cómo crear un esquema GraphQL que nos permita consultar los datos de varios clientes y aplicaciones.

Empezando con TDEngine

La forma más sencilla de empezar a utilizar TDEngine es utilizar su servicio en la nube. Vaya a TDEngine y cree una cuenta. Tienen algunas bases de datos públicas que podemos usar, lo cual es excelente para armar una demostración o experimentar con consultas.

Si desea ejecutar TDEngine localmente, puede utilizar la imagen de Docker y Telegraf para recuperar datos de varias fuentes y enviarlos a la base de datos, como información del sistema, estadísticas de ping, etc.

 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

Consulte la documentación oficial para la configuración de Telegraf y la documentación de TDEngine en Telegraf . En resumen, esto se vería así para conectarse a un tema MQTT:

 [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", ]

En lugar de configurar todo localmente y esperar a que la base de datos se llene de información, usaremos la base de datos pública para este artículo, que contiene los movimientos de barcos de los 5 principales puertos de EE. UU.

Uso de TDEngine con datos públicos de movimiento de barcos

De forma predeterminada, las tablas en TDEngine tienen un esquema implícito, lo que significa que el esquema se adapta a los datos que se escriben en la base de datos. Esto es excelente para el arranque, pero eventualmente queremos cambiar a un esquema explícito para evitar problemas con los datos entrantes. Una cosa a la que lleva un poco de tiempo acostumbrarse es a su concepto de Super Tables ("STable" para abreviar). En TDEngine hay etiquetas (claves) y columnas (datos). Para cada combinación de teclas, se crea una "tabla". Todas las tablas están agrupadas en STable.

Al observar la base de datos vessel , tienen una STable llamada ais_data que contiene muchas tablas. Por lo general, no queremos realizar consultas por tabla, sino que siempre usamos STable para obtener datos acumulados de todas las tablas.

TDEngine tiene una función DESCRIBE que nos permite inspeccionar el esquema de una tabla o STable. ais_data tiene el siguiente esquema:

STable tiene dos claves y seis columnas de datos. Las claves son el mmsi y el name . Podemos usar declaraciones SQL regulares para consultar los datos:

 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 ...

Tenga en cuenta que los datos de series temporales suelen ser muy grandes, por lo que siempre debemos limitar el conjunto de resultados. Hay algunas funciones específicas de series temporales que podemos usar, como PARTITION BY , que agrupa los resultados por clave y es útil para obtener las últimas actualizaciones de claves individuales. Por ejemplo:

 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 ... 

Recomiendo leer su documentación SQL para obtener más ejemplos. Antes de continuar, dirígete a "Programación", "Node.js" y recupera tus variables TDENGINE_CLOUD_URL y TDENGINE_CLOUD_TOKEN .

GraphQL con Nexus.js, Fastify y Mercurius

GraphQL es bastante conocido hoy en día y hay muchos buenos artículos al respecto. Elegimos la tecnología porque recopilamos y procesamos información de diferentes fuentes y GraphQL nos permite combinarlas en una única API de forma transparente.

Usaremos el increíble marco Fastify (ahora la opción predeterminada para las aplicaciones Node.js) y el adaptador Mercurius . Los equipos de Mercurius y Fastify trabajaron juntos para lograr una experiencia perfecta y es una excelente elección de API GraphQL con enfoque en el rendimiento. GraphQL Nexus es una herramienta para construir/generar el esquema y los solucionadores, por lo que no tenemos que escribir todo a mano.

Hay un poco de código de configuración, etc. por hacer, que omitiré aquí. Puede encontrar un ejemplo completo en GitHub: tdengine-graphql-example .

Quiero profundizar en dos cosas en este artículo que son bastante específicas:

1. la biblioteca de consultas TDEngine
2. el esquema GraphQL con Nexus

Biblioteca de consultas TDEngine

TDEngine cuenta con una biblioteca Node.js que nos permite consultar la base de datos. Esto facilita la conexión y el envío de consultas; desafortunadamente, es un poco difícil trabajar con las respuestas. Entonces escribimos un pequeño envoltorio:

 '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 }) }

Esto devuelve un objeto TDEngine que se puede pasar al contexto GraphQL. Principalmente usaremos la función fetchData donde podemos pasar una consulta SQL y obtener los resultados como una matriz de objetos. TDEngine devuelve los metadatos (columnas), errores y datos por separado. Usaremos los metadatos para asignar las columnas a una lista normal de objetos. Un caso especial aquí es la función last_row . Las columnas se devuelven como last_row(ts) , last_row(name) etc. y queremos eliminar la parte last_row para que el atributo se asigne 1:1 al esquema GraphQL. Esto se hace en la parte columnName.replace .

Esquema GraphQL

Desafortunadamente, no existe un generador de esquemas como Postgraphile para TDEngine y no queremos escribir ni mantener un esquema GraphQL puro, por lo que usaremos Nexus.js para ayudarnos con eso. Comenzaremos con dos tipos básicos: VesselMovement y Timestamp (que es un tipo escalar). Timestamp y TDDate son dos tipos diferentes para mostrar la fecha como una marca de tiempo o como una cadena de fecha. Esto es útil para la aplicación cliente (y durante el desarrollo), ya que puede decidir qué formato usar. asNexusMethod nos permite usar el tipo como una función en el esquema VesselMovement . Podemos resolver TDDate aquí mismo en la definición de tipo para usar el valor de marca de tiempo ts original.

 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') } })

Para los tipos de series temporales, utilizamos el sufijo Movement o Series para una separación clara de los tipos relacionales y de series temporales en la interfaz.

Ahora podemos definir la Consulta. Comenzaremos con una consulta simple para obtener los últimos movimientos de TDEngine:

 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 es una gran herramienta para probar la API y explorar el esquema; puede habilitarla pasando graphiql.enabled = true en Mercurius. Con la consulta podemos ver los últimos movimientos de embarcaciones agrupados por mmsi . Aunque vayamos un poco más allá. Una de las mayores ventajas de GraphQL es que es una capa transparente para el cliente o la aplicación. Podemos recuperar datos de múltiples fuentes y combinarlos en el mismo esquema.

Desafortunadamente, no pude encontrar una API fácil y gratuita con información detallada sobre los buques. Existe Sinay , pero solo proporcionan el name , mmsi e imo en su respuesta Vessel (que ya tenemos en TDEngine). Por el bien del ejemplo, asumimos que no tenemos el name en nuestra base de datos y necesitamos recuperarlo de Sinay. Con imo también podríamos consultar las emisiones de CO2 de un buque u otra API podría usarse para recuperar una imagen, la bandera u otra información, todo lo cual se puede combinar en el tipo Vessel .

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

Como puede ver aquí, podemos incluir una lista movements de campos con los datos de series temporales de TDEngine. Agregaremos otra consulta para obtener la información del barco y el solucionador nos permitirá combinar los datos de TDEngine y Sinay:

 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] } } }) 

🎉 y aquí tenemos una API GraphQL en funcionamiento que devuelve filas de TDEngine para un recipiente específico que solicitamos. getVesselInformation() es un contenedor simple para recuperar datos de Sinay. Agregaremos los resultados de TDEngine al atributo movements y GraphQL se encargará del resto y asignará todo al esquema.

Nota: Inyección SQL

Como ocurre con cualquier base de datos SQL, debemos tener cuidado con la entrada del usuario. En el ejemplo anterior utilizamos la entrada mmsi directamente, lo que hace que esta consulta sea vulnerable a las inyecciones de SQL. Por el bien del ejemplo, ignoraremos esto por ahora, pero en aplicaciones del "mundo real", siempre debemos desinfectar la entrada del usuario. Existen varias bibliotecas pequeñas para desinfectar cadenas; en la mayoría de los casos, solo confiamos en números (paginación, límite, etc.) y enumeraciones (orden de clasificación), que GraphQL verifica por nosotros.

¡Gracias a Dmitry Zaets por señalar esto!

Optimizaciones

Hay algunas cosas que van más allá del alcance de este artículo, pero quiero mencionarlas brevemente:

Pothos como sucesor espiritual de Nexus.js

Cuando iniciamos el proyecto, Nexus.js era la mejor opción para generar nuestro esquema GraphQL. Aunque es estable y tiene algunas funciones completas , carece de mantenimiento y actualizaciones. Existe un generador de esquemas GraphQL basado en complementos llamado Pothos , que es un poco más moderno y se mantiene activamente. Si está iniciando un nuevo proyecto, probablemente le recomiendo usar Pothos en lugar de Nexus.js.

¡Gracias a Mo Sattler por señalar esto!

Resolutores de campo

Como puede ver en el solucionador Vessel anterior, ambas fuentes de datos se obtienen y procesan inmediatamente. Esto significa que si la consulta es solo para el name , aún obtenemos los movements para la respuesta. Y si la consulta es solo para los movements , igualmente obtenemos el nombre de Sinay y potencialmente pagamos por la solicitud.

Ese es un antipatrón de GraphQL y podemos mejorar el rendimiento utilizando la información del campo para recuperar solo los datos que se solicitan. Los resolutores tienen la información de campo como cuarto argumento, pero es bastante difícil trabajar con ellos. En su lugar, podemos usar graphql-parse-fields para obtener un objeto simple de los campos solicitados y ajustar la lógica de resolución.

Optimizaciones de consultas SQL

En nuestras consultas de ejemplo, usamos select * para recuperar todas las columnas de la base de datos incluso si no son necesarias. Obviamente, esto es bastante malo y podemos usar el mismo analizador de campos para optimizar las consultas SQL:

 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(',') }

Esta función devuelve una lista de campos separados por comas de la información de GraphQL.

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

Si solicitamos ts , latitude y longitude , la consulta se vería así:

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

Con solo unas pocas columnas en esta tabla, esto puede no importar mucho, pero con más tablas y consultas complejas, esto puede marcar una gran diferencia en el rendimiento de la aplicación.

Funciones de series temporales

TDEngine tiene algunas extensiones específicas de series temporales que deberían usarse para mejorar el rendimiento. Por ejemplo, para recuperar la última entrada, una consulta SQL tradicional:

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

Tarda 653 ms en ejecutarse, mientras que la consulta "TDEngine" tarda sólo 145 ms:

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

Hay opciones de configuración para cada tabla para optimizar las funciones de última fila/primera fila y otras configuraciones de caché. Recomiendo leer la documentación de TDEngine .

Conclusión

La versión simple: en este artículo, configuramos una base de datos de series temporales de TDEngine y definimos un esquema GraphQL para permitir que las aplicaciones cliente se conecten y consulten datos.

Hay mucho más. Tenemos un proyecto estándar para combinar datos complejos de series de tiempo con datos relacionales en una interfaz transparente. En Nevados, utilizamos PostgreSQL como base de datos principal y recuperamos datos de series de tiempo de la misma manera que en el ejemplo movement anterior. Esta es una excelente manera de combinar datos de múltiples fuentes en una sola API. Otro beneficio es que los datos solo se obtienen cuando se solicitan, lo que agrega mucha flexibilidad a la aplicación cliente. Por último, pero no menos importante, el esquema GraphQL funciona como documentación y contrato, por lo que podemos marcar fácilmente la casilla "Documentación API".

Si tiene alguna pregunta o comentario , comuníquese con BlueSky o únase a la discusión en GitHub .