EMQX Enterprise

日本語

简体中文
English

日本語

简体中文
English

Appearance

お問い合わせ
お問い合わせ

MQTTデータをTablestoreに取り込む ​

Tablestoreは、IoTシナリオに最適化されたスケーラブルでサーバレスなデータベースです。時系列データ、構造化データ、半構造化データを管理するためのワンストップソリューションであるIoTstoreを提供しています。IoT、車載ネットワーク、リスクコントロール、メッセージング、レコメンデーションシステムなどのシナリオに最適です。Tablestoreは、コスト効率が高く高性能なデータストレージを提供し、ミリ秒単位のクエリや取得、柔軟なデータ分析機能を備えています。EMQXはTablestore Cloud、Tablestore OSS、Tablestore Enterpriseとシームレスに統合し、IoTユースケースにおける効率的なデータ管理を実現します。

動作概要 ​

EMQXにおけるTablestoreデータ統合は、EMQXのリアルタイムデータキャプチャと送信機能と、Tablestoreの高性能なデータストレージおよび分析機能をシームレスに組み合わせています。組み込みのルールエンジンを活用することで、EMQXからTablestoreへのデータ取り込みと保存のプロセスを簡素化し、複雑なコーディングを不要にします。EMQXはルールエンジンとSinkを介してIoTデバイスのデータをTablestoreに転送し、効率的な保存と分析を可能にします。

データが保存されると、Tablestoreはレポートやチャート、その他の可視化を生成する強力なツールを提供し、これらはTablestoreの可視化機能を通じてユーザーに提示されます。

以下の図は、エネルギー貯蔵シナリオにおけるEMQXとTablestoreの典型的なデータ統合アーキテクチャを示しています。

MQTT to Tablestore

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モデルを対象としています。

前提条件 ​

以下を事前にご確認ください。

時系列テーブルの作成 ​

  1. Tablestoreコンソールにログインします。
  2. 時系列モデルのインスタンスを作成します。インスタンス名として例としてemqx-demoを指定します。インスタンス作成の詳細はTablestore公式ドキュメントを参照してください。
  3. インスタンス管理ページに移動します。
  4. インスタンス詳細タブで時系列テーブルを選択し、時系列テーブル作成ボタンをクリックします。
  5. 時系列テーブル情報を設定し、テーブル名に例としてtimeseries_demo_with_dataを入力します。確認をクリックします。

img

時系列テーブルの管理 ​

先ほど作成した時系列テーブルを管理するには、テーブル名をクリックして時系列テーブル管理ページに入ります。ビジネス要件に応じて以下の手順を実行できます。

  1. データクエリタブをクリックします。

  2. 時系列追加をクリックします。

    TIP

    この手順は任意です。時系列テーブルが存在しない場合、データ書き込み時にTablestoreが自動的に作成します。そのため、この例では時系列の手動操作は示していません。

img

コネクターの作成 ​

このセクションでは、SinkをTablestoreサーバーに接続するためのコネクター作成方法を示します。

以下の手順はEMQXとTablestoreをローカルマシンで実行していることを前提としています。リモート環境で実行している場合は設定を適宜調整してください。

  1. EMQXダッシュボードに入り、Integration -> Connectorsをクリックします。
  2. ページ右上の作成をクリックします。
  3. コネクター作成ページでTablestoreを選択し、次へをクリックします。
  4. 設定ステップで以下を設定します。
    • コネクター名を入力します。英数字の組み合わせで指定してください。例：my_tablestore
    • Tablestoreサーバー接続情報を入力します。
      • エンドポイント：TablestoreインスタンスのアクセスURLを入力します。Tablestoreコンソールのインスタンス詳細ページで確認可能です。デプロイ方法に応じてURLを入力してください。例：パブリックネットワークの場合はhttps://emqx-demo.cn-hangzhou.ots.aliyuncs.com
      • インスタンス名：接続するTablestoreインスタンス名。例として先ほど作成したemqx-demoを使用します。
      • Access Key ID：Tablestore認証に使用するAlibaba Cloud発行のアクセスキーID。
      • Access Key Secret：アクセスキーIDに対応するシークレットキー。
      • ストレージモデルタイプ：現時点ではTimeSeriesのみサポート。
      • プローブテーブル名：EMQX 6.3.2以降、既存の時系列テーブルを指定して接続性チェックに利用可能です。多くの時系列テーブルがあるインスタンスで推奨されます。未設定の場合、EMQXはエンドポイントの生のTCP接続性をプローブし、Tablestoreの認証やテーブルアクセスは検証しません。
    • TLSパラメータの設定。TablestoreはHTTPSエンドポイントを使用するため、TLSはデフォルトで有効です。追加のTLS設定は不要です。TLS接続オプションの詳細は外部リソースアクセスのTLS有効化を参照してください。
  5. 作成をクリックする前に、接続テストをクリックしてコネクターがTablestoreに接続できるか確認できます。
  6. ページ下部の作成ボタンをクリックしてコネクター作成を完了します。ポップアップでコネクター一覧に戻るかルール作成を選択できます。ルールとSinkの作成手順はTablestore Sink付きルールの作成を参照してください。

Tablestore Sink付きルールの作成 ​

このセクションでは、EMQXでソースMQTTトピックt/#からのメッセージを処理し、設定済みのSinkを通じてTablestoreに送信するルールの作成方法を示します。

  1. EMQXダッシュボードに入り、左メニューからIntegration -> Rulesをクリックします。

  2. ページ右上の作成をクリックします。

  3. ルール作成ページで、ルールIDにmy_ruleを入力します。

  4. SQLエディターでルールを設定します。例えば、トピックt/#のMQTTメッセージをTablestoreに保存したい場合、以下のSQL文を使用します。

    TIP

    独自のSQL文を指定する場合は、後で設定するSinkのデータ形式に含まれるすべての変数がSELECT部分に含まれていることを確認してください。

    sql
    SELECT
  *
FROM
  "t/#"

    注：初心者の方はSQL例をクリックし、テスト有効化でSQLルールの学習とテストが可能です。

    • アクション追加ボタンをクリックし、ルールがトリガーするアクションを定義します。このアクションでEMQXはルールで処理したデータをTablestoreに送信します。

  6. アクションの種類ドロップダウンからAlibaba Tablestoreを選択します。アクションはデフォルトのCreate Actionのままにします。既に作成済みのSinkがあれば選択可能です。この例では新規Sinkを作成します。

  7. Sinkの名前を入力します。英数字の組み合わせで指定してください。

  8. コネクタードロップダウンから先ほど作成したmy_tablestoreを選択します。新規コネクターはドロップダウン横のボタンから作成可能です。設定パラメータはコネクターの作成を参照してください。

  9. 以下の項目を設定します。

    • データソース：EMQXがメッセージを取得するデータソース。処理対象のデータの起点を示します。特定のトピックやデータストリームを指定します。

    • テーブル名：データを保存するTablestoreのテーブル名。先ほど作成したテーブル名を入力します。${table}などの変数で動的に割り当てることも可能です。

    • メジャメント：Tablestoreで使用されるメジャメント名。通常はデータの論理的なグループやカテゴリを示します。例：temperature_readingsやsensor_data。${measurement}などの変数も使用可能です。

    • ストレージモデルタイプ：Tablestoreで使用するデータストレージモデルの種類。現時点ではtimeseriesのみサポートし、時系列データに最適化されています。

    • タグ：Tablestoreの各データエントリに関連付けるキー・バリュー形式のタグ。メタデータやラベルとして利用し、クエリやフィルタリングを容易にします。追加をクリックして複数定義可能です。例：

      キー値
      locationoffice1
      devicesensor1

    • フィールド：Tablestoreに送信するデータフィールドのリスト。各フィールドはTablestoreテーブルのカラムにマッピングされます。追加をクリックして以下を設定します。

      • カラム：Tablestoreのカラム名。${column_name}などの変数で定義可能で、後述のメッセージペイロードのフィールドと一致させます。
      • メッセージ値：カラムに割り当てる値。${value}のような動的参照、trueのような真偽値、1.3の数値、バイナリデータなどが指定可能です。
      • 整数型か：カラムが数値型の場合、EMQXはデフォルトで浮動小数点型としてTablestoreに挿入します。整数値として挿入したい場合はこのフラグをtrueに設定します。設定ファイル経由では${isint}などの変数で動的割り当ても可能です。
      • バイナリ型か：カラムがバイナリの場合、EMQXはデフォルトで文字列型として挿入します。バイナリデータとして挿入したい場合はこのフラグをtrueに設定します。設定ファイル経由では${isbinary}などの変数で動的割り当ても可能です。

    • タイムスタンプ：Tablestoreに記録されるタイムスタンプ。マイクロ秒単位の整数値で指定します。固定値、文字列"NOW"（EMQXがメッセージ処理時に現在時刻を動的に埋め込み）、${microsecond_timestamp}のような変数プレースホルダーで動的割り当てが可能です。

    • メタ更新モデル：Tablestoreのメタデータ更新戦略を定義します。

      • MUM_IGNORE：メタデータ更新を無視し、競合があってもメタデータを変更しません。
      • MUM_NORMAL：通常のメタデータ更新を行います。メタデータが存在しない場合は動的に作成してからデータを書き込みます。既存メタデータと競合があれば上書きされる可能性があります。

  10. フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。これらはプライマリSinkがメッセージ処理に失敗した場合にトリガーされます。詳細はフォールバックアクションを参照してください。

  11. 詳細設定（任意）：詳細設定を参照してください。

  12. 作成をクリックする前に、接続テストをクリックしてSinkがTablestoreに接続できるか確認できます。

  13. 作成をクリックしてSink作成を完了します。ルール作成ページのアクション出力タブに新しいSinkが表示されます。

  14. ルール作成ページで設定内容を確認し、作成ボタンをクリックしてルールを生成します。

これでルールが正常に作成され、ルールページに新しいルールが表示されます。**アクション(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コンソールにアクセスし、データがTablestoreに書き込まれているか確認します。

    • メトリック名にメジャメント名（このデモではfoo）を入力。
    • タグにlocation=office1とdevice=sensor1をクエリ条件として入力し、検索をクリック。

    tablestore_query_data

詳細設定 ​

このセクションでは、TablestoreコネクターおよびSinkの詳細設定オプションについて説明します。ダッシュボードでコネクターやSinkを設定する際、詳細設定に進み、以下のパラメータをニーズに合わせて調整できます。

項目説明推奨値
バッファプールサイズEMQXとTablestore間の出口タイプブリッジでデータフローを管理するために割り当てられるバッファワーカープロセスの数を指定します。これらのワーカーは送信前のデータを一時的に保存・処理します。出口（アウトバウンド）シナリオのパフォーマンス最適化に重要です。入力（インバウンド）データのみを扱うSinkの場合は"0"に設定可能です。16
リクエストTTLリクエストTTL（Time To Live）は、リクエストがバッファに入ってから有効と見なされる最大秒数を指定します。TTLを超えてバッファに滞留するか、送信後にTablestoreからの応答やアックがタイムリーに得られない場合、リクエストは期限切れと判断されます。45
ヘルスチェック間隔SinkがTablestore接続の自動ヘルスチェックを実行する間隔（秒）を指定します。15
最大バッファキューサイズTablestore Sinkの各バッファワーカーがバッファリング可能な最大バイト数を指定します。バッファワーカーはデータ送信前に一時的にデータを保持し、データフローを効率化します。システムの性能やデータ転送要件に応じて調整してください。256
バッチサイズEMQXからTablestoreに一度に転送可能なデータバッチのサイズを指定します。サイズを調整することでデータ転送の効率と性能を最適化できます。1
クエリモードメッセージ送信を最適化するためにasynchronous（非同期）またはsynchronous（同期）モードを選択できます。非同期モードではTablestoreへの書き込みがMQTTメッセージのパブリッシュをブロックしませんが、クライアントがTablestore到着前にメッセージを受信する可能性があります。Async
インフライトウィンドウインフライトクエリとは、開始されたがまだ応答やアックを受け取っていないクエリを指します。SinkがTablestoreと通信する際に同時に存在可能な最大インフライトクエリ数を制御します。
クエリモードがasync（非同期）の場合、このパラメータは特に重要です。同一MQTTクライアントからのメッセージを厳密な順序で処理したい場合は、この値を1に設定してください。		100