Criando um banco de dados de série temporal com TDEngine e GraphQL

Motivação e Introdução

Como parte da equipe de software da Nevados, estamos construindo uma plataforma de operações e monitoramento para o Nevados All Terrain Tracker®. Um rastreador solar é um dispositivo que orienta um painel solar em direção ao sol. Todo rastreador solar envia constantemente informações de status e leituras, como ângulo atual, temperatura, tensões, etc. para nossa plataforma e precisamos armazenar essas informações para análise e visualização. Se o rastreador estiver configurado para enviar dados a cada 5 segundos, teremos 17.280 pontos de dados por rastreador por dia, 518.400 pontos de dados por rastreador por mês. Isso resume muita informação. Este tipo de dados é denominado "dados de séries temporais" e como para todos os problemas complexos de software, existem diversas soluções (bancos de dados de séries temporais) para eles. Os mais famosos são InfluxDB e TimescaleDB. Para nossa plataforma, decidimos trabalhar com o TDEngine , um produto relativamente novo que é otimizado para aplicações IoT e funciona com a linguagem de consulta SQL.

Houve vários argumentos para esta decisão: TDEngine

• é de código aberto
• é otimizado para aplicativos IoT
• usa SQL, que é uma linguagem com a qual estamos familiarizados
• está disponível como um serviço gerenciado e podemos nos concentrar na construção de nosso aplicativo
• é fácil de executar localmente via Docker

Neste artigo, veremos a configuração de um banco de dados e tabelas TDEngine e como criar um esquema GraphQL que nos permita consultar os dados de vários clientes e aplicativos.

Introdução ao TDEngine

A maneira mais fácil de começar a usar o TDEngine é usar o serviço em nuvem. Acesse o TDEngine e crie uma conta. Eles têm alguns bancos de dados públicos que podemos usar, o que é ótimo para montar uma demonstração ou experimentar consultas.

Se quiser executar o TDEngine localmente, você pode usar a imagem Docker e o Telegraf para recuperar dados de várias fontes e enviá-los ao banco de dados, como informações do sistema, estatí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

Confira a documentação oficial para a configuração do Telegraf e a documentação do TDEngine no Telegraf . Resumindo, seria mais ou menos assim para conectar-se a um tópico 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", ]

Em vez de configurar tudo localmente e esperar que o banco de dados seja preenchido com informações, usaremos o banco de dados público para este artigo, que contém movimentos de navios dos 5 principais portos dos EUA.

Usando TDEngine com dados públicos de movimento de navios

Por padrão, as tabelas no TDEngine possuem um esquema implícito, o que significa que o esquema se adapta aos dados que são gravados no banco de dados. Isso é ótimo para inicialização, mas, eventualmente, queremos mudar para um esquema explícito para evitar problemas com os dados recebidos. Uma coisa com a qual leva um pouco de tempo para se acostumar é o conceito de Super Tables ("STable", para abreviar). No TDEngine existem tags (chaves) e colunas (dados). Para cada combinação de teclas, é criada uma “tabela”. Todas as tabelas são agrupadas no STable.

Olhando para o banco de dados vessel , eles têm um STable chamado ais_data que contém muitas tabelas. Normalmente, não queremos consultar por tabela, mas sempre usamos STable para obter dados acumulados de todas as tabelas.

TDEngine possui uma função DESCRIBE que nos permite inspecionar o esquema de uma tabela ou STable. O ais_data possui o seguinte esquema:

O STable possui duas chaves e seis colunas de dados. As chaves são o mmsi e o name . Podemos usar instruções SQL regulares para consultar os dados:

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

Tenha em mente que os dados de séries temporais geralmente são muito grandes, portanto devemos sempre limitar o conjunto de resultados. Existem algumas funções específicas de séries temporais que podemos usar, como PARTITION BY , que agrupa os resultados por chave e é útil para obter as chaves individuais de atualização mais recentes. Por exemplo:

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

Recomendo a leitura da documentação SQL para mais exemplos. Antes de prosseguirmos, vá para "Programação", "Node.js" e recupere suas variáveis TDENGINE_CLOUD_URL e TDENGINE_CLOUD_TOKEN .

GraphQL com Nexus.js, Fastify e Mercurius

GraphQL é bastante conhecido atualmente e há muitos artigos bons sobre ele. Escolhemos a tecnologia porque coletamos e processamos informações de diferentes fontes e o GraphQL nos permite combiná-las em uma única API de forma transparente.

Usaremos a incrível estrutura Fastify (agora a escolha padrão para aplicativos Node.js) e o adaptador Mercurius . As equipes Mercurius e Fastify trabalharam juntas para uma experiência perfeita e é uma ótima escolha de APIs GraphQL com foco em desempenho. GraphQL Nexus é uma ferramenta para construir/gerar o esquema e os resolvedores, portanto não precisamos escrever tudo manualmente.

Há um pouco de código de configuração etc. a ser feito, que vou pular aqui. Você pode encontrar um exemplo completo no GitHub - tdengine-graphql-example .

Quero elaborar duas coisas neste artigo que são bastante específicas:

1. a biblioteca de consulta TDEngine
2. o esquema GraphQL com Nexus

Biblioteca de consulta TDEngine

TDEngine possui uma biblioteca Node.js que nos permite consultar o banco de dados. Isso facilita a conexão e o envio de consultas, infelizmente as respostas são um pouco difíceis de trabalhar. Então escrevemos um pequeno wrapper:

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

Isso retorna um objeto TDEngine que pode ser passado para o contexto GraphQL. Usaremos principalmente a função fetchData , onde podemos passar uma consulta SQL e obter os resultados como uma matriz de objetos. TDEngine retorna os metadados (colunas), erros e dados separadamente. Usaremos os metadados para mapear as colunas em uma lista regular de objetos. Um caso especial aqui é a função last_row . As colunas são retornadas como last_row(ts) , last_row(name) etc. e queremos remover a parte last_row para que o atributo mapeie 1:1 para o esquema GraphQL. Isso é feito na parte columnName.replace .

Esquema GraphQL

Infelizmente não existe um gerador de esquema como Postgraphile para TDEngine e não queremos escrever e manter um esquema GraphQL puro, então usaremos Nexus.js para nos ajudar com isso. Começaremos com dois tipos básicos: VesselMovement e Timestamp (que é um tipo escalar). Timestamp e TDDate são dois tipos diferentes para exibir a data como carimbo de data/hora ou como string de data. Isto é útil para a aplicação cliente (e durante o desenvolvimento), pois pode decidir qual formato usar. asNexusMethod nos permite usar o tipo como uma função no esquema VesselMovement . Podemos resolver o TDDate aqui mesmo na definição do tipo para usar o valor ts timestamp 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 tipos de série temporal, usamos o sufixo Movement ou Series para uma separação clara dos tipos relacionais e de série temporal na interface.

Agora podemos definir a Consulta. Começaremos com uma consulta simples para obter os movimentos mais recentes do 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 é uma ótima ferramenta para testar a API e explorar o esquema, você pode habilitá-lo passando graphiql.enabled = true no Mercurius. Com a consulta podemos ver os últimos movimentos das embarcações agrupadas por mmsi . Vamos um pouco mais longe. Uma das maiores vantagens do GraphQL é que ele é uma camada transparente para o cliente ou aplicação. Podemos buscar dados de múltiplas fontes e combiná-los no mesmo esquema.

Infelizmente, não consegui encontrar uma API fácil/gratuita com informações extensas sobre a embarcação. Existe o Sinay , mas eles fornecem apenas o name , mmsi e imo na resposta do Vessel (que já temos no TDEngine). Para fins de exemplo, presumimos que não temos o name em nosso banco de dados e precisamos recuperá-lo do Sinay. Com o imo também poderíamos consultar as emissões de CO2 de uma embarcação ou outra API poderia ser usada para recuperar uma imagem, a bandeira ou outras informações, todas as quais podem ser combinadas no 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 você pode ver aqui, podemos incluir uma lista de movements de campo com os dados de série temporal do TDEngine. Adicionaremos outra consulta para buscar as informações da embarcação e o resolvedor nos permitirá combinar os dados do TDEngine e do 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] } } }) 

🎉 e aqui temos uma API GraphQL funcional retornando linhas do TDEngine para um navio específico que solicitamos. getVesselInformation() é um wrapper simples para buscar dados do Sinay. Adicionaremos os resultados do TDEngine ao atributo movements e o GraphQL cuidará do resto e mapeará tudo para o esquema.

Nota: Injeção SQL

Como acontece com qualquer banco de dados SQL, precisamos ter cuidado com as entradas do usuário. No exemplo acima usamos a entrada mmsi diretamente, o que torna esta consulta vulnerável a injeções de SQL. Para fins de exemplo, vamos ignorar isso por enquanto, mas em aplicações do “mundo real”, devemos sempre higienizar a entrada do usuário. Existem várias pequenas bibliotecas para higienizar strings, na maioria dos casos contamos apenas com números (paginação, limite etc.) e enums (ordem de classificação), que o GraphQL verifica para nós.

Obrigado a Dmitry Zaets por apontar isso!

Otimizações

Existem algumas coisas que vão além do escopo deste artigo, mas quero mencioná-las brevemente:

Pothos como sucessor espiritual do Nexus.js

Quando iniciamos o projeto, Nexus.js foi a melhor escolha para gerar nosso esquema GraphQL. Embora estável e com alguns recursos completos , falta manutenção e atualizações. Existe um construtor de esquema GraphQL baseado em plug-in chamado Pothos , que é um pouco mais moderno e mantido ativamente. Se você estiver iniciando um novo projeto, provavelmente recomendo usar Pothos em vez de Nexus.js.

Obrigado a Mo Sattler por apontar isso!

Resolvedores de campo

Como você pode ver no resolvedor Vessel acima, ambas as fontes de dados são imediatamente buscadas e processadas. Isso significa que se a consulta for apenas para name , ainda buscaremos os movements para a resposta. E se a consulta for apenas para movements , ainda buscaremos o nome do Sinay e potencialmente pagaremos pela solicitação.

Esse é um antipadrão GraphQL e podemos melhorar o desempenho usando as informações do campo para buscar apenas os dados solicitados. Os resolvedores têm as informações do campo como quarto argumento, mas são muito difíceis de trabalhar. Em vez disso, podemos usar graphql-parse-fields para obter um objeto simples dos campos solicitados e ajustar a lógica do resolvedor.

Otimizações de consulta SQL

Em nossas consultas de exemplo, usamos select * para buscar todas as colunas do banco de dados, mesmo que não sejam necessárias. Obviamente, isso é muito ruim e podemos usar o mesmo analisador de campo para otimizar as 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 função retorna uma lista de campos separados por vírgula das informações do GraphQL.

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

Se solicitarmos ts , latitude e longitude , a consulta ficaria assim:

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

Com apenas algumas colunas nesta tabela isso pode não importar muito, mas com mais tabelas e consultas complexas, isso pode fazer uma enorme diferença no desempenho do aplicativo.

Funções de série temporal

O TDEngine possui algumas extensões específicas de série temporal que devem ser usadas para melhorar o desempenho. Por exemplo, para recuperar a entrada mais recente, uma consulta SQL tradicional:

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

Leva 653ms para ser executada, enquanto a consulta "TDEngine" leva apenas 145ms:

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

Existem opções de configuração para cada tabela para otimizar as funções last_row/first_row e outras configurações de cache. Recomendo a leitura da documentação do TDEngine .

Conclusão

A versão simples: neste artigo, configuramos um banco de dados de série temporal TDEngine e definimos um esquema GraphQL para permitir que aplicativos clientes se conectem e consultem dados.

Há muito mais do que isso. Temos um projeto padrão para combinar dados complexos de séries temporais com dados relacionais em uma interface transparente. Em Nevados, estamos usando PostgreSQL como banco de dados primário e recuperamos dados de série temporal da mesma forma que no exemplo movement acima. Esta é uma ótima maneira de combinar dados de múltiplas fontes em uma única API. Outro benefício é que os dados só são buscados quando solicitados, o que agrega muita flexibilidade à aplicação cliente. Por último, mas não menos importante, o esquema GraphQL funciona como documentação e contrato, para que possamos facilmente marcar a caixa “Documentação da API”.

Se você tiver alguma dúvida ou comentário , entre em contato pelo BlueSky ou participe da discussão no GitHub .