# MQTTデータをTablestoreに取り込む [Tablestore](https://www.alibabacloud.com/en/product/table-store/pricing?spm=a3c0i.29367734.6737026690.8.78847d3fcEhuVv&_p_lc=1)は、IoTシナリオに最適化されたスケーラブルでサーバーレスなデータベースです。時系列データ、構造化データ、半構造化データを管理するためのワンストップソリューションであるIoTstoreを提供しています。IoT、車載ネットワーク、リスク管理、メッセージング、レコメンデーションシステムなどのシナリオに最適です。Tablestoreは、コスト効率が高く高性能なデータストレージを提供し、ミリ秒単位のクエリや検索、柔軟なデータ分析機能を備えています。EMQXはTablestore Cloud、Tablestore OSS、Tablestore Enterpriseとシームレスに統合し、IoTユースケースにおける効率的なデータ管理を実現します。 ## 動作概要 EMQXにおけるTablestoreのデータ統合は、EMQXのリアルタイムデータキャプチャと転送機能と、Tablestoreの高性能なデータストレージおよび分析機能をシームレスに組み合わせたものです。組み込みの[ルールエンジン](./rules.md)を活用することで、EMQXからTablestoreへのデータ取り込みと保存のプロセスを簡素化し、複雑なコーディングを不要にします。EMQXはルールエンジンとSinkを通じてIoTデバイスのデータをTablestoreに転送し、効率的な保存と分析を可能にします。 データが保存されると、Tablestoreはレポートやチャート、その他の可視化を生成する強力な分析ツールを提供し、これらはTablestoreの可視化機能を通じてユーザーに提示されます。 以下の図は、エネルギー貯蔵シナリオにおけるEMQXとTablestore間の典型的なデータ統合アーキテクチャを示しています。 ![MQTT to Tablestore](./assets/mqtt-to-tablestore.png) EMQXとTablestoreは、エネルギー消費データをリアルタイムに効率的に収集・分析するための拡張可能なIoTプラットフォームを提供します。このアーキテクチャでは、EMQXがIoTプラットフォームとしてデバイスのアクセス管理、メッセージの送受信、データルーティングを担当し、Tablestoreがデータストレージおよび分析プラットフォームとしてデータの保存と分析機能を担います。ワークフローは以下の通りです。 1. **メッセージのパブリッシュと受信**：エネルギー貯蔵デバイスや産業用IoTデバイスはMQTTプロトコルを通じてEMQXに正常に接続し、電力消費量、入出力電力などの情報を含むエネルギー消費データを定期的にMQTTプロトコルでパブリッシュします。EMQXはこれらのメッセージを受信すると、ルールエンジン内でマッチング処理を開始します。 2. **メッセージデータの処理**：組み込みのルールエンジンを使用して、特定のソースからのメッセージをトピックマッチングに基づいて処理できます。メッセージが到着するとルールエンジンを通過し、対応するルールとマッチングしてメッセージデータを処理します。例えば、データ形式の変換、特定情報のフィルタリング、コンテキスト情報によるメッセージの付加などです。 3. **Tablestoreへのデータ取り込み**：ルールエンジンで定義されたルールがトリガーとなり、メッセージをTablestoreに書き込む操作が実行されます。Tablestore Sinkは設定可能なフィールドを提供し、書き込むデータ形式を柔軟に定義でき、メッセージの特定フィールドをTablestoreの対応するメジャメントやフィールドにマッピングします。 エネルギー消費データがTablestoreに書き込まれた後、以下のような分析が可能です。 - Grafanaなどの可視化ツールに接続し、データに基づくチャートを生成してエネルギー貯蔵データを表示する。 - 業務システムに接続して、エネルギー貯蔵デバイスの状態監視やアラートを行う。 ## 特長と利点 Tablestoreデータ統合は以下の特長と利点を提供します。 - **効率的なデータ処理**：EMQXは大量のIoTデバイス接続とメッセージスループットを処理でき、Tablestoreはデータの書き込み、保存、クエリに優れた性能を発揮します。IoTシナリオのデータ処理ニーズに対応しつつ、システムに過度な負荷をかけません。 - **メッセージ変換**：EMQXのルールを通じて、メッセージはTablestoreに書き込まれる前に多様な処理や変換を受けることができます。 - **スケーラビリティ**：EMQXとTablestoreはどちらもクラスター拡張に対応しており、ビジネスの成長に応じてクラスターを柔軟に水平拡張できます。 - **豊富なクエリ機能**：Tablestoreは最適化された関数、演算子、インデックス技術を提供し、時系列データの効率的なクエリと分析を可能にし、IoT時系列データから価値ある洞察を正確に抽出します。 - **効率的なストレージ**：Tablestoreは高圧縮率のエンコーディング方式を採用し、ストレージコストを大幅に削減します。また、異なるデータタイプに対して保存期間をカスタマイズでき、不必要なデータがストレージを占有するのを防ぎます。 ## はじめる前に このセクションでは、Tablestoreデータ統合の作成を開始する前に必要な準備、すなわちデータベースインスタンスの作成や時系列テーブルの作成・管理について説明します。 ::: tip 現時点では、Tablestoreとのデータ統合はTimeSeriesモデルのみをサポートしています。したがって、以下の手順はTimeSeriesモデルに焦点を当てています。 ::: ### 前提条件 作業を進める前に、以下を確認してください。 - EMQXのデータ統合[ルール](./rules.md)の理解。 - EMQXにおける[データ統合](./data-bridges.md)の仕組みの知識。 ### 時系列テーブルの作成 1. [Tablestoreコンソール](https://account.alibabacloud.com/login/login.htm)にログインします。 2. 時系列モデルのインスタンスを作成します。インスタンス名を例えば`emqx-demo`とします。インスタンス作成の詳細は[Tablestore公式ドキュメント](https://www.alibabacloud.com/help/en/tablestore/product-overview/activate-service-and-create-a-instance)を参照してください。 3. **インスタンス管理**ページに移動します。 4. **インスタンス詳細**タブで**時系列テーブル**を選択し、**時系列テーブルの作成**ボタンをクリックします。 5. 時系列テーブル情報を設定し、テーブル名を例えば`timeseries_demo_with_data`と入力して**確認**をクリックします。 ![img](./assets/tablestore_instance_manage.png) ### 時系列テーブルの管理 先ほど作成した時系列テーブルを管理するには、テーブル名をクリックして**時系列テーブル管理**ページに入ります。ここから、ビジネス要件に応じて以下の操作が可能です。 1. **データクエリ**タブをクリックします。 2. **時系列の追加**をクリックします。 ::: tip このステップは任意です。時系列テーブルがまだ存在しない場合、データ書き込み時にTablestoreが自動的に作成します。したがって、この例では時系列の手動操作は示していません。 ::: ![img](./assets/tablestore_timeline_mamge.png) ## コネクターの作成 このセクションでは、SinkをTablestoreサーバーに接続するためのコネクター作成方法を示します。 以下の手順は、EMQXとTablestoreの両方をローカルマシンで実行していることを前提としています。リモートで実行している場合は設定を適宜調整してください。 1. EMQXダッシュボードに入り、**Integration** -> **Connectors**をクリックします。 2. ページ右上の**Create**をクリックします。 3. **Create Connector**ページで**Tablestore**を選択し、**Next**をクリックします。 4. **Configuration**ステップで以下を設定します： - コネクター名を入力します。英数字の組み合わせで、例：`my_tablestore`。 - Tablestoreサーバー接続情報を入力します： - **Endpoint**：TablestoreインスタンスのアクセスURLを入力します。Tablestoreコンソールのインスタンス詳細ページで確認可能です。例えば、パブリックネットワークの場合は`https://emqx-demo.cn-hangzhou.ots.aliyuncs.com`など。 - **Instance Name**：接続するTablestoreインスタンス名。例：`emqx-demo`。 - **Access Key ID**：Tablestore認証に使用するAccess Key ID。Alibaba Cloudが発行するキーです。 - **Access Key Secret**：Access Key IDに紐づく認証用のAccess Key Secret。 - **Storage Model Type**：現在は`TimeSeries`のみサポート。 - TLSパラメータの設定。TablestoreはHTTPSエンドポイントを使用するため、TLSはデフォルトで有効です。追加設定は不要です。TLS接続オプションの詳細は[外部リソースアクセスのTLS有効化](../../guides/network/overview.md#enabling-tls-for-external-resource-access)を参照してください。 5. **Create**をクリックする前に、**Test Connectivity**でコネクターがTablestoreサーバーに接続できるかテスト可能です。 6. ページ下部の**Create**をクリックしてコネクター作成を完了します。ポップアップで**Back to Connector List**または**Create Rule**を選択可能です。ルールとSinkを作成してTablestoreへ転送するデータを指定する場合は、[Tablestore Sinkを使ったルール作成](#create-a-rule-with-tablestore-sink)を参照してください。 ## Tablestore Sinkを使ったルール作成 このセクションでは、EMQXでソースMQTTトピック`t/#`からのメッセージを処理し、設定したSinkを通じてTablestoreに送信するルールの作成方法を示します。 1. EMQXダッシュボードに入り、左ナビゲーションメニューから**Integration** -> **Rules**をクリックします。 2. ページ右上の**Create**をクリックします。 3. ルール作成ページで、ルールIDに`my_rule`を入力します。 4. **SQL Editor**でルールを設定します。例えば、トピック`t/#`のMQTTメッセージをTablestoreに保存したい場合、以下のSQL文を使用します。 ::: tip 独自のSQL文を指定する場合、後で設定するSinkのデータ形式に含まれるすべての変数が`SELECT`句に含まれていることを確認してください。 ::: ```sql SELECT * FROM "t/#" ``` 注：初心者の場合は、**SQL Examples**をクリックし、**Enable Test**でSQLルールを学習・テストできます。 5. + **Add Action**ボタンをクリックして、ルールがトリガーするアクションを定義します。このアクションにより、EMQXはルールで処理したデータをTablestoreに送信します。 6. **Type of Action**ドロップダウンから`Alibaba Tablestore`を選択します。**Action**はデフォルトの`Create Action`のままにします。既に作成済みのSinkがあれば選択可能です。この例では新規Sinkを作成します。 7. Sinkの名前を入力します。英数字の組み合わせで指定してください。 8. **Connector**ドロップダウンから先ほど作成した`my_tablestore`を選択します。隣のボタンから新規コネクター作成も可能です。設定パラメータは[コネクターの作成](#create-a-connector)を参照してください。 9. 以下のフィールドを設定します： - **Data Source**：EMQXがメッセージを取得するデータソース。処理対象のデータの起点を示します。特定のトピックやデータストリームを指定します。 - **Table Name**：データを保存するTablestoreのテーブル名。先に作成したテーブル名を入力します。`${table}`などの変数を使って動的に割り当てることも可能です。 - **Measurement**：Tablestoreで使用するメジャメント名。通常はデータの論理的なグループやカテゴリを示します。例：`temperature_readings`や`sensor_data`。`${measurement}`などの変数も使用可能です。 - **Storage Model Type**：Tablestoreで使用するデータストレージモデルのタイプ。現在は`timeseries`のみサポートされ、時系列データに最適化されています。 - **Tags**：Tablestoreの各データエントリーに関連付けるキー・バリュー形式のタグ。メタデータやラベルとして利用し、クエリやフィルタリングを容易にします。**Add**をクリックして複数のタグを定義可能です。例： | Key | Value | | ---------- | --------- | | `location` | `office1` | | `device` | `sensor1` | - **Fields**：Tablestoreに送信するデータのフィールドリスト。各フィールドはTablestoreテーブルのカラムにマッピングされます。**Add**をクリックして以下を追加します： - **Column**：Tablestoreのカラム名。`${column_name}`などの変数を使って定義可能で、後述のメッセージペイロードのフィールドと対応させます。 - **Message value**：カラムに割り当てる値。`${value}`のような動的参照、真偽値（`true`）、数値（`1.3`）、バイナリデータなどが指定可能です。 - **Is Int**：カラムが数値型の場合、EMQXはデフォルトで浮動小数点型としてTablestoreに挿入します。整数値として挿入したい場合はこのフラグを`true`に設定します。設定ファイル経由では`${isint}`のような変数で動的に割り当て可能です。 - **Is Binary**：カラムがバイナリ型の場合、EMQXはデフォルトで文字列型として挿入します。バイナリデータとして挿入したい場合はこのフラグを`true`に設定します。設定ファイル経由では`${isbinary}`のような変数で動的に割り当て可能です。 - **Timestamp**：Tablestoreに記録されるタイムスタンプ。マイクロ秒単位の整数値で指定します。固定値、文字列"NOW"（EMQXがメッセージ処理時に現在時刻を動的に埋め込み）、`${microsecond_timestamp}`のような変数プレースホルダーも使用可能です。 - **Meta Update Model**：Tablestoreのメタデータ更新戦略を定義します： - `MUM_IGNORE`：メタデータ更新を無視し、競合があってもメタデータを変更しません。 - `MUM_NORMAL`：通常のメタデータ更新を行います。メタデータが存在しない場合は動的に作成し、既存メタデータと競合があれば上書きされる可能性があります。 10. **フォールバックアクション（任意）**：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。これらはプライマリSinkがメッセージ処理に失敗した場合にトリガーされます。詳細は[フォールバックアクション](./data-bridges.md#fallback-actions)を参照してください。 11. **詳細設定（任意）**：[詳細設定](#advanced-configurations)を参照してください。 12. **Create**をクリックする前に、**Test Connectivity**でSinkがTablestoreサーバーに接続可能かテストできます。 13. **Create**をクリックしてSink作成を完了します。ルール作成ページに戻ると、**Action Outputs**タブに新しいSinkが表示されます。 14. ルール作成ページで設定内容を確認し、**Create**ボタンをクリックしてルールを生成します。 これでルールが正常に作成され、**Rule**ページに新しいルールが表示されます。**Actions(Sink)**タブをクリックすると、新しいTablestore Sinkが確認できます。 また、**Integration** -> **Flow Designer**をクリックするとトポロジーが表示され、トピック`t/#`のメッセージがルール`my_rule`で解析されてTablestoreに送信・保存されている様子が確認できます。 ## ルールのテスト 1. MQTTXを使ってトピック`t/1`にメッセージを送信し、オンライン/オフラインイベントをトリガーします。 ```bash mqttx pub -i emqx_c -t t/1 -m '{ "table": "timeseries_demo_with_data", "measurement": "foo", "microsecond_timestamp": 1734924039271024, "column_name": "cc", "value": 1}' ``` 2. Sinkの稼働状況を確認し、新しい受信メッセージと送信メッセージがそれぞれ1件ずつあることを確認します。 3. [Tablestoreコンソール](https://account.alibabacloud.com/login/login.htm?spm=5176.12901015-2.0.0.1a364b84fgwsH6)にアクセスし、データがTablestoreに書き込まれているか確認します。 - **Metric Name**にメジャメント名（このデモでは`foo`）を入力します。 - **Tag**に`location=office1`および`device=sensor1`をクエリ条件として入力し、**Search**をクリックします。 ![tablestore_query_data](./assets/tablestore_query_data.png) ## 詳細設定 このセクションでは、TablestoreコネクターおよびSinkの詳細設定オプションについて説明します。ダッシュボードでコネクターやSinkを設定する際、**Advanced Settings**に進み、以下のパラメータをニーズに合わせて調整できます。 | **項目** | **説明** | **推奨値** | | ---------------------- | ---------------------------------------------------------------------------------------------------------------- | ---------- | | Buffer Pool Size | EMQXとTablestore間の出口タイプのブリッジでデータフロー管理に割り当てるバッファワーカープロセス数を指定します。これらのワーカーは、ターゲットサービスに送信する前のデータを一時的に保存・処理します。出口（アウトバウンド）シナリオのパフォーマンス最適化やスムーズなデータ送信に重要です。Ingress（インバウンド）データのみ扱うSinkの場合は"0"に設定可能です。 | `16` | | Request TTL | リクエストTTL（Time To Live）は、リクエストがバッファに入ってから有効とみなされる最大秒数を指定します。TTLを超えてバッファに滞留するか、送信後にTablestoreからの応答やアックがタイムリーに得られない場合、リクエストは期限切れと見なされます。 | `45` | | Health Check Interval | SinkがTablestoreとの接続状態を自動的にヘルスチェックする間隔（秒）を指定します。 | `15` | | Max Buffer Queue Size | Tablestore Sinkの各バッファワーカーがバッファリング可能な最大バイト数を指定します。バッファワーカーはデータを一時的に保存し、効率的なデータフローを実現します。システム性能やデータ転送要件に応じて調整してください。 | `256` | | Batch Size | EMQXからTablestoreへ一度に転送可能なデータバッチのサイズを指定します。サイズ調整によりデータ転送の効率と性能を微調整できます。 | `1` | | Query Mode | メッセージ送信の最適化のため、`asynchronous`（非同期）または`synchronous`（同期）モードを選択可能です。非同期モードではTablestoreへの書き込みがMQTTメッセージのパブリッシュ処理をブロックしませんが、クライアントがTablestore到着前にメッセージを受信する可能性があります。 | `Async` | | Inflight Window | 「インフライトクエリ」とは、開始されたがまだ応答やアックを受け取っていないクエリを指します。SinkがTablestoreと通信する際に同時に存在できる最大インフライトクエリ数を制御します。
**Query Mode**が`async`の場合、このパラメータは特に重要です。同一MQTTクライアントからのメッセージを厳密に順序処理したい場合は、値を1に設定してください。 | `100` |