TDEngine と GraphQL を使用した時系列データベースの作成

動機と紹介

ネバドスのソフトウェア チームの一員として、私たちは Nevados All Terrain Tracker® の運用および監視プラットフォームを構築しています。ソーラートラッカーは、ソーラーパネルを太陽の方向に向ける装置です。すべてのソーラー トラッカーは、現在の角度、温度、電圧などのステータス情報と測定値をプラットフォームに常に送信しており、分析と視覚化のためにこの情報を保存する必要があります。トラッカーが 5 秒ごとにデータを送信するように構成されている場合、1 日あたりトラッカーあたり 17,280 データ ポイント、1 か月あたりトラッカーあたり 518,400 データ ポイントになります。これで多くの情報が要約されます。この種のデータは「時系列データ」と呼ばれ、ソフトウェアにおけるすべての複雑な問題については、いくつかの解決策 (時系列データベース) があります。最も有名なものは InfluxDB と TimescaleDB です。私たちのプラットフォームでは、IoT アプリケーション用に最適化され SQL クエリ言語で動作する比較的新しい製品であるTDEngineを使用することにしました。

この決定にはいくつかの議論がありました: TDEngine

• オープンソースです
• IoTアプリケーション向けに最適化されています
• 私たちが使い慣れた言語である SQL を使用します
• マネージド サービスとして利用できるため、アプリケーションの構築に集中できます
• Docker経由でローカルで簡単に実行できます

この記事では、TDEngine データベースとテーブルのセットアップと、さまざまなクライアントやアプリケーションからデータをクエリできるようにする GraphQL スキーマの作成方法について説明します。

TDEngine を始める

TDEngine を使い始める最も簡単な方法は、クラウド サービスを使用することです。 TDEngineにアクセスしてアカウントを作成します。彼らは私たちが使用できる公開データベースをいくつか持っているので、デモをまとめたり、クエリを実験したりするのに最適です。

TDEngine をローカルで実行する場合は、Docker イメージとTelegraf を使用して、システム情報、ping 統計などのデータをさまざまなソースから取得し、データベースに送信できます。

 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

Telegraf 構成の公式ドキュメントとTelegraf の TDEngine ドキュメントを確認してください。つまり、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", ]

すべてをローカルで設定してデータベースに情報が入力されるのを待つ代わりに、この記事では、米国の 5 つの主要な港からの船舶の移動が含まれるパブリック データベースを使用します。

TDEngine と公開船舶移動データの使用

デフォルトでは、TDEngine のテーブルには暗黙的なスキーマがあり、データベースに書き込まれるデータにスキーマが適応することを意味します。これはブートストラップには最適ですが、最終的には、受信データの問題を回避するために明示的なスキーマに切り替えたいと考えています。慣れるまでに少し時間がかかることの 1 つは、スーパー テーブル(略して「STable」) の概念です。 TDEngine にはタグ (キー) と列 (データ) があります。キーの組み合わせごとに「テーブル」が作成されます。すべてのテーブルは STable にグループ化されます。

vesselデータベースを見ると、 ais_dataという名前の STable が 1 つあり、これには多くのテーブルが含まれています。通常、テーブルごとにクエリを実行する必要はありませんが、すべてのテーブルから蓄積されたデータを取得するには常に STable を使用します。

TDEngine には、テーブルまたは STable のスキーマを検査できる関数DESCRIBEがあります。 ais_dataには次のスキーマがあります。

STable には 2 つのキーと 6 つのデータ列があります。キーはmmsiとnameです。通常の SQL ステートメントを使用してデータをクエリできます。

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

時系列データは通常非常に大きいため、常に結果セットを制限する必要があることに注意してください。 PARTITION BYは結果をキーごとにグループ化し、最新の更新の個々のキーを取得するのに役立ちます。例えば：

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

その他の例については、 SQL ドキュメントを読むことをお勧めします。次に進む前に、「プログラミング」、「Node.js」に移動し、 TDENGINE_CLOUD_URL変数とTDENGINE_CLOUD_TOKEN変数を取得します。

Nexus.js、Fastify、Mercurius を使用した GraphQL

GraphQL は最近ではかなりよく知られており、それに関する優れた記事がたくさんあります。さまざまなソースから情報を収集して処理するため、このテクノロジーを選択しました。GraphQL を使用すると、それらを単一の API に透過的に組み合わせることができます。

ここでは、素晴らしいFastifyフレームワーク (現在では Node.js アプリケーションのデフォルトの選択肢となっています) とMercuriusアダプターを使用します。 Mercurius と Fastify のチームはシームレスなエクスペリエンスを実現するために協力しており、パフォーマンスを重視した GraphQL API としては最適です。 GraphQL Nexusはスキーマとリゾルバーを構築/生成するツールなので、すべてを手作業で記述する必要はありません。

セットアップ コードなどが少し必要ですが、ここでは省略します。完全な例はGitHub - tdengine-graphql-exampleで見つけることができます。

この記事では、かなり具体的な 2 つのことについて詳しく説明したいと思います。

1. TDEngine クエリ ライブラリ
2. Nexus を使用した GraphQL スキーマ

TDEngine クエリ ライブラリ

TDEngine には、データベースにクエリを実行できるNode.js ライブラリがあります。これにより、接続とクエリの送信が簡単になりますが、残念ながら、応答の処理が少し難しくなります。そこで、小さなラッパーを書きました。

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

これにより、GraphQL コンテキストに渡すことができる TDEngine オブジェクトが返されます。主に、SQL クエリを渡して結果をオブジェクトの配列として取得できるfetchData関数を使用します。 TDEngine は、メタデータ (列)、エラー、データを個別に返します。メタデータを使用して、列をオブジェクトの通常のリストにマップします。ここでの特殊なケースは、 last_row関数です。列はlast_row(ts) 、 last_row(name)などとして返されます。属性が GraphQL スキーマに 1:1 でマップされるように、 last_row部分を削除したいと考えています。これは、 columnName.replace部分で行われます。

GraphQL スキーマ

残念ながら、TDEngine にはPostgraphileのようなスキーマ ジェネレーターはなく、純粋な GraphQL スキーマを作成して維持する必要はないため、Nexus.js を使用してそれを支援します。まず、 VesselMovementとTimestamp (スカラー タイプ) という 2 つの基本タイプから始めます。 TimestampとTDDate日付をタイムスタンプまたは日付文字列として表示する 2 つの異なるタイプです。これは、使用する形式を決定できるため、クライアント アプリケーション (および開発中) に役立ちます。 asNexusMethod使用すると、型をVesselMovementスキーマの関数として使用できます。ここで型定義内でTDDate解決して、元のtsタイムスタンプ値を使用できます。

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

時系列タイプの場合、インターフェイス内でリレーショナル タイプと時系列タイプを明確に分離するために、 MovementまたはSeriesサフィックスを使用します。

これでクエリを定義できるようになりました。 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 は、API をテストし、スキーマを探索するための優れたツールです。Mercurius でgraphiql.enabled = trueを渡すことで有効にできます。このクエリを使用すると、 mmsiにグループ化された船舶の最新の動きを確認できます。ただし、もう少し先に進みましょう。 GraphQL の最大の利点の 1 つは、クライアントまたはアプリケーションにとって透過的なレイヤーであることです。複数のソースからデータをフェッチし、それらを同じスキーマに結合できます。

残念ながら、広範な船舶情報を備えた簡単で無料の API は見つかりませんでした。 Sinayがありますが、Vessel 応答 (TDEngine にはすでにあります) でname 、 mmsi 、およびimoのみが提供されます。例として、データベースにnameがなく、Sinay から名前を取得する必要があると仮定します。 imoを使用すると、船舶の CO2 排出量をクエリしたり、別の API を使用して画像、旗、その他の情報を取得したりすることもでき、これらすべてを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' }) } })

ここでわかるように、TDEngine からの時系列データにリスト フィールドのmovements含めることができます。船舶情報を取得するための別のクエリを追加します。リゾルバーを使用すると、TDEngine と 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] } } }) 

🎉 そして、ここには、リクエストした特定の船舶について TDEngine から行を返す、動作する GraphQL API があります。 getVesselInformation() Sinay からデータをフェッチするための単純なラッパーです。 TDEngine の結果をmovements属性に追加すると、GraphQL が残りを処理してすべてをスキーマにマッピングします。

注: SQL インジェクション

他の SQL データベースと同様に、ユーザー入力には注意する必要があります。上の例ではmmsi入力を直接使用しているため、このクエリは SQL インジェクションに対して脆弱になります。例のために、今はこれを無視しますが、「現実世界」のアプリケーションでは、常にユーザー入力をサニタイズする必要があります。文字列をサニタイズするための小さなライブラリがいくつかありますが、ほとんどの場合、GraphQL がチェックする数値 (ページネーション、制限など) と列挙型 (ソート順) のみに依存します。

これを指摘してくれた Dmitry Zaets に感謝します。

最適化

この記事の範囲を超える内容がいくつかありますが、簡単に説明したいと思います。

Nexus.js の精神的な後継者としての Pothos

プロジェクトを開始したとき、GraphQL スキーマを生成するには Nexus.js が最良の選択でした。安定していて機能もある程度充実していますが、メンテナンスやアップデートが不足しています。 Pothosと呼ばれるプラグインベースの GraphQL スキーマ ビルダーがあり、これはもう少し最新で積極的にメンテナンスされています。新しいプロジェクトを開始する場合は、Nexus.js の代わりに Pothos を使用することをお勧めします。

これを指摘してくれた Mo Sattler に感謝します。

フィールドリゾルバー

上記のVesselリゾルバーでわかるように、両方のデータ ソースがすぐにフェッチされて処理されます。これは、クエリがnameのみに対するものである場合でも、応答のmovementsを取得することを意味します。また、クエリがmovementsのみに関するものである場合でも、Sinay から名前を取得し、リクエストに対する料金を支払う可能性があります。

これは GraphQL のアンチパターンであり、フィールド情報を使用して要求されたデータのみをフェッチすることでパフォーマンスを向上させることができます。リゾルバーには 4 番目の引数としてフィールド情報がありますが、これを扱うのは非常に困難です。代わりに、 graphql-parse-fields使用して、要求されたフィールドの単純なオブジェクトを取得し、リゾルバー ロジックを調整できます。

SQLクエリの最適化

このクエリ例では、 select *を使用して、必要でない場合でもデータベースからすべての列を取得します。これは明らかにかなり悪いものであり、同じフィールド パーサーを使用して 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(',') }

この関数は、GraphQL 情報からフィールドのカンマ区切りリストを返します。

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

ts 、 latitude 、 longitudeリクエストすると、クエリは次のようになります。

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

このテーブルに列が数個しかない場合、これはあまり問題にならないかもしれませんが、より多くのテーブルと複雑なクエリを使用すると、アプリケーションのパフォーマンスに大きな違いが生じる可能性があります。

時系列関数

TDEngine には、パフォーマンスを向上させるために使用する必要がある時系列固有の拡張機能がいくつかあります。たとえば、最新のエントリを取得するには、従来の SQL クエリを使用します。

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

実行には 653 ミリ秒かかりますが、「TDEngine」クエリには 145 ミリ秒しかかかりません。

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

各テーブルには、last_row/first_row 関数やその他のキャッシュ設定を最適化するための構成オプションがあります。 TDEngine のドキュメントを読むことをお勧めします。

結論

単純なバージョン: この記事では、TDEngine 時系列データベースをセットアップし、クライアント アプリケーションがデータに接続してクエリできるように GraphQL スキーマを定義しました。

他にもたくさんあります。複雑な時系列データとリレーショナル データを透過的なインターフェイスで組み合わせる定型プロジェクトがあります。 Nevados では、PostgreSQL をプライマリ データベースとして使用し、上記のmovement例と同じ方法で時系列データを取得しています。これは、複数のソースからのデータを 1 つの API に結合する優れた方法です。もう 1 つの利点は、要求された場合にのみデータがフェッチされるため、クライアント アプリケーションに大幅な柔軟性が追加されることです。最後になりましたが、GraphQL スキーマはドキュメントおよび契約として機能するため、[API ドキュメント] ボックスに簡単にチェックを入れることができます。

ご質問やコメントがございましたら、 BlueSky にご連絡いただくか、 GitHub のディスカッションに参加してください。