EMQX Enterprise

日本語

简体中文
English

日本語

简体中文
English

Appearance

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

MongoDBへのMQTTデータ取り込み ​

MongoDBは、スキーマ設計の柔軟性、スケーラビリティ、大量の構造化および半構造化データの保存能力で知られる主要なNoSQLデータベースです。EMQXとMongoDBを統合することで、ユーザーはMQTTメッセージやクライアントイベントを直接MongoDBに効率的に取り込むことができます。これにより、MongoDB内での長期的な時系列データの保存や高度なクエリ機能が可能になります。この統合は一方向のデータフローを保証し、EMQXからのMQTTメッセージがMongoDBデータベースに書き込まれます。この強力な組み合わせは、IoTデータを効果的に管理したい企業にとって堅実な基盤となります。

本ページでは、EMQXとMongoDB間のデータ統合について包括的に紹介し、データ統合の作成および検証に関する実践的な手順を提供します。

動作概要 ​

MongoDBデータ統合は、MQTTベースのIoTデータとMongoDBの強力なデータ保存機能をつなぐためにEMQXに標準搭載された機能です。組み込みのルールエンジンコンポーネントを利用することで、EMQXからMongoDBへのデータ取り込みを簡素化し、複雑なコーディングを不要にします。

以下の図は、EMQXとMongoDB間のデータ統合の典型的なアーキテクチャを示しています。

mongdb_bridge_architecture

MongoDBへのMQTTデータ取り込みは以下のように動作します：

  1. メッセージのパブリッシュと受信：接続された車両、IIoTシステム、エネルギー管理プラットフォームなどのIoTデバイスは、MQTTプロトコルを介してEMQXに正常に接続し、特定のトピックにMQTTメッセージをパブリッシュします。EMQXがこれらのメッセージを受信すると、ルールエンジン内でマッチング処理を開始します。
  2. メッセージデータの処理：メッセージが到着すると、ルールエンジンを通過し、EMQXで定義されたルールにより処理されます。ルールは事前定義された条件に基づき、どのメッセージをMongoDBにルーティングするかを決定します。ペイロード変換が指定されている場合は、データ形式の変換、特定情報のフィルタリング、追加コンテキストによるペイロードの強化などの変換が適用されます。
  3. MongoDBへのデータ取り込み：ルールエンジンがMongoDB保存対象のメッセージを特定すると、MongoDBへの転送アクションがトリガーされます。処理済みデータはMongoDBデータベースのコレクションにシームレスに書き込まれます。
  4. データの保存と活用：データがMongoDBに保存されることで、企業はそのクエリ機能を活用して様々なユースケースに対応できます。例えば、接続車両の分野では、保存されたデータを用いて車両の健康状態を把握したり、リアルタイムの指標に基づくルート最適化や資産追跡が可能です。同様にIIoT環境では、機械の状態監視、メンテナンス予測、生産スケジュールの最適化に活用されます。

この統合システムを利用することで、電力・エネルギー分野の企業はグリッドの健康状態を継続的に監視し、需要予測や潜在的な障害の早期発見が可能になります。リアルタイムおよび履歴データから得られる価値は、運用効率の向上だけでなく、コスト削減や顧客体験の向上にもつながります。

特長とメリット ​

MongoDBとのデータ統合は、効果的なデータ処理と保存を保証するための多様な特長とメリットを提供します：

  • IoTデータ管理の効率化

    IoTデータの取り込み、保存、処理、分析を一元化し、複雑な統合や面倒なデータ移行を排除します。データサイロを解消し、IoTデータの統合ビューを実現します。

  • リアルタイムデータ処理

    EMQXはリアルタイムデータストリームの処理に最適化されており、ソースシステムからMongoDBへの効率的かつ信頼性の高いデータ伝送を保証します。即時のインサイトやアクションが必要なユースケースに理想的です。

  • 柔軟なMongoDB接続オプション

    単一のMongoDBインスタンスでも、レプリカセットの堅牢性を活用する場合でも、両方の構成にネイティブ対応しており、インフラ要件に応じて柔軟に適応可能です。

  • 高性能かつスケーラブル

    EMQXの分散アーキテクチャとMongoDBのカラムナストレージ形式により、データ量の増加に伴いシームレスにスケール可能です。大規模データセットでも一貫したパフォーマンスと応答性を維持します。IoT展開の拡大に合わせてデータ保存能力を容易に拡張できます。

  • 柔軟なデータ変換

    EMQXは強力なSQLベースのルールエンジンを提供し、MongoDBに保存する前にデータを前処理できます。フィルタリング、ルーティング、集約、強化など多様な変換機能をサポートし、ニーズに応じたデータ整形が可能です。

  • NoSQLの利点

    MongoDBのスキーマレスアーキテクチャにより、多様なMQTTメッセージ構造を厳格なスキーマなしで容易に保存でき、IoTデータの動的な性質に対応します。

  • 信頼性の高いデータ保存

    EMQXのルールエンジンがメッセージを処理・ルーティングした後、MongoDBに保存され、プラットフォームの実績ある信頼性によりデータの整合性と継続的な可用性が保証されます。

  • 運用指標と高度な分析

    総メッセージ数、送信トラフィックレートなどの指標からインサイトを得られます。これらの指標とMongoDBの強力なクエリ機能を組み合わせることで、データフローの監視、分析、最適化が可能となり、予測分析や異常検知などに活用できます。

  • 最新のMongoDBバージョン対応

    データ統合は最新のMongoDBバージョンに対応しており、最新機能、最適化、セキュリティアップデートの恩恵を受けられます。

  • コスト効率

    EMQXとMongoDBは共にオープンソースソリューションであり、プロプライエタリなソリューションと比較してコスト効率に優れています。これにより、IoTプロジェクトの総所有コスト削減と投資収益率の向上に寄与します。

このMongoDBデータ統合は、IoTインフラを強化し、デバイスから生成される膨大なデータを単に保存するだけでなく、将来のクエリや分析に備えて準備します。セットアップの容易さと運用の優秀性により、IoTシステムの効率性と信頼性を大幅に向上させます。

はじめる前に ​

このセクションでは、EMQXダッシュボードでMongoDBデータ統合を作成する前に完了すべき準備について説明します。

前提条件 ​

MongoDBサーバーのセットアップ ​

以下のコマンドを使用して、Docker経由でMongoDBをインストールし、Dockerイメージを起動し、ユーザーを作成できます。

bash
# MongoDBのDockerイメージを起動し、パスワードをpublicに設定
docker run -d --name mongodb -p 27017:27017 mongo

# コンテナにアクセス
docker exec -it mongodb bash

# コンテナ内でMongoDBサーバーに接続（4.xバージョンでは`mongosh`を使用）
mongosh

# ユーザー作成
use admin
db.createUser({ user: "admin", pwd: "public", roles: [ { role: "root", db: "admin" } ] })

データベースの作成 ​

以下のコマンドでMongoDBにデータベースとコレクションを作成できます。

bash
# データベース emqx_data を作成
use emqx_data

# コレクション emqx_messages を作成
db.createCollection('emqx_messages')

コネクターの作成 ​

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

以下の手順は、EMQXとMongoDBの両方をローカルマシンで実行していることを前提としています。MongoDBが別の場所にデプロイされている場合は、設定を適宜調整してください。

  1. EMQXダッシュボードに入り、Integration -> Connectorsをクリックします。

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

  3. Create ConnectorページでMongoDBを選択し、Nextをクリックします。

  4. コネクターの名前を入力します。名前は大文字・小文字の英数字の組み合わせにしてください。例：my_mongodb

  5. MongoDBサーバーの接続情報を設定します。必須項目（＊印）を入力してください。

    • MongoDB Mode：実際のデプロイモードに基づき接続するMongoDBのタイプを選択します。この例ではsingleを選択します。
      • single：単一のスタンドアロンMongoDBインスタンス
      • rs：同じデータセットを維持するmongodプロセスのレプリカセット
      • sharded：MongoDBのシャーディングクラスター
    • Server Host：127.0.0.1:27017またはMongoDBサーバーがリモートの場合は実際のURLを入力
    • Database Name：emqx_dataを入力
    • Write Mode：デフォルトのunsafeのまま
    • Username：adminを入力
    • Password：publicを入力
    • Auth Source：ユーザー認証に使用するデータベース名を入力
    • Use Legacy Protocol：MongoDBのレガシー通信プロトコルを使用するかどうかを設定（MongoDB 3.6以降は新しいワイヤープロトコルを導入、レガシープロトコルは後方互換用）。true、false、autoから選択可能。auto（デフォルト）では、検出されたMongoDBバージョンに基づき自動判別します。
    • Srv Record：デフォルトで無効。有効にすると、DNS SRVレコードを使用して接続すべきMongoDBホストを検出でき、レプリカセットやシャーディングクラスターへの接続が容易になります。
    • 暗号化接続を確立したい場合は、Enable TLSのトグルスイッチをオンにします。TLS接続の詳細は外部リソースアクセスのTLSを参照してください。

  6. フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のために、1つ以上のフォールバックアクションを定義できます。詳細はフォールバックアクションを参照してください。

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

  8. Createをクリックする前に、Test ConnectivityをクリックしてコネクターがMongoDBサーバーに接続できるかテストできます。

  9. ページ下部のCreateボタンをクリックしてコネクター作成を完了します。ポップアップダイアログでBack to Connector Listをクリックするか、Create RuleをクリックしてルールとSinkの作成を続行できます。詳しい手順はルールとMongoDB Sinkの作成を参照してください。

MongoDB Sinkを用いたルールの作成 ​

このセクションでは、ダッシュボードでMQTTトピックt/#からのメッセージを処理し、処理済みデータを設定済みのMongoDB Sink経由でMongoDBに保存するルールの作成方法を示します。

  1. EMQXダッシュボードでIntegration -> Rulesをクリックします。

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

  3. ルールIDにmy_ruleを入力し、SQL Editorでルールを設定します。トピックt/#のMQTTメッセージをMongoDBに保存したい場合、以下のSQL構文を使用できます。

    注意：独自のSQL構文を指定する場合は、Sinkが必要とするすべてのフィールドをSELECT句に含めてください。

    sql
    SELECT
  *
FROM
  "t/#"

    例えば、timestampを日付型として保存し、payloadをJSON文字列として保存するには以下のSQLを使用します：

    sql
    SELECT
  *,
  mongo_date(timestamp) as timestamp,
  json_encode(payload) as payload
FROM
  "t/#"

    注意：初心者の方はSQL Examplesをクリックし、Enable Testを有効にしてSQLルールを学習・テストしてください。

    • Add Actionボタンをクリックし、ルールによってトリガーされるアクションを定義します。このアクションにより、EMQXはルールで処理したデータをMongoDBに送信します。

  5. Type of ActionドロップダウンリストからMongoDBを選択します。ActionドロップダウンはデフォルトのCreate Actionのままにします。既に作成済みのSinkを選択することも可能です。この例では新しいSinkを作成します。

  6. Sinkの名前を入力します。名前は大文字・小文字の英数字の組み合わせにしてください。

  7. Connectorドロップダウンからmy_mongodbを選択します。隣のボタンをクリックして新しいコネクターを作成することも可能です。設定パラメータの詳細はコネクターの作成を参照してください。

  8. Collectionフィールドにデータを保存するコレクション名を入力します。${var_name}のプレースホルダーを使った動的設定もサポートしています。この例ではemqx_messagesを入力します。

  9. Payload templateを設定し、clientid、topic、qos、timestamp、payloadをMongoDBに保存します。このテンプレートはMongoDBのinsertコマンドで実行され、サンプルコードは以下の通りです：

    json
    {
  "clientid": "${clientid}",
  "topic": "${topic}",
  "qos": ${qos},
  "timestamp": ${timestamp},
  "payload": ${payload}
}

    ペイロードテンプレート設定時の注意点：

    • すべてのkeyはダブルクォーテーション"で囲む必要があります。

    • 値のデータ型の自動判別はサポートされていません：

      • 文字列は"で囲む必要があります。囲まないとエラーになります。
      • 数値などの値は囲まないでください。囲むと文字列として認識されます。
      • timestamp、日付、時間型は特別な処理をしないと数値または文字列として扱われます。日付や時間として保存するには、ルールSQLのMongoDBタイム関数を使用してください。（注：これらの関数の戻り値はMongoDBアクションでのみ有効で、他のアクションでは使用できません。）

    • 値がJSONオブジェクトの場合はネストされたオブジェクトが許容されます：

      • テンプレート内で値を"で囲んでネストすることはできません。実行エラーになります。
      • オブジェクトはその構造のままネストされて保存されます。

    • オブジェクトをJSON文字列として保存したい場合は、ルールSQLのjson_encode関数で変換し、テンプレート内の対応する値は"で囲まないでください。

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

  11. Createをクリックする前に、Test ConnectivityをクリックしてSinkがMongoDBサーバーに接続できるかテストできます。

  12. CreateボタンをクリックしてSink設定を完了します。新しいSinkがAction Outputsに追加されます。

  13. Create Ruleページに戻り、設定内容を確認します。Createボタンをクリックしてルールを生成します。

これでルールが正常に作成され、Ruleページに新しいルールが表示されます。**Actions(Sink)**タブをクリックすると、新しいMongoDB Sinkが確認できます。

また、Integration -> Flow Designerをクリックするとトポロジーが表示され、トピックt/#のメッセージがルールmy_ruleで解析されMongoDBに送信・保存されていることが確認できます。

ルールのテスト ​

ルールとSinkが期待通りに動作するかテストするために、MQTTXを使ってクライアントをシミュレートし、EMQXにMQTTメッセージをパブリッシュできます。

  1. MQTTXを使ってトピックt/1にメッセージを送信します：

    bash
    mqttx pub -i emqx_c -t t/1 -m '{ "msg": "hello MongoDB" }'

  2. Sinkの稼働状況を確認し、Matchedが1件、Sent Successfullyが1件増えていることを確認します。

  3. メッセージがコレクションemqx_messagesに書き込まれているか確認します：

    > db.emqx_messages.find().pretty()
{
    "_id" : ObjectId("63db7059df489d01ed000009"),
    "clientid" : "emqx_c",
    "payload" : {
      "msg" : "hello MongoDB"
    },
    "qos" : 0,
    "timestamp" : NumberLong("1675325529070"),
    "topic" : "t/1"
}

    ルール設定で2番目のSQL構文を使用している場合、返される情報は以下のようになります：

    > db.emqx_messages.find().pretty()
{
    "_id" : ObjectId("63db7535df489d01ed000013"),
    "clientid" : "emqx_c",
    "payload" : "{ \"msg\": \"hello MongoDB\" }",
    "qos" : 0,
    "timestamp" : ISODate("2023-02-02T08:33:36.715Z"),
    "topic" : "t/1"
}

詳細設定 ​

このセクションでは、EMQX MongoDBコネクターおよびSinkの詳細設定オプションについて紹介します。コネクターやSinkの設定時にAdvanced Settingsを展開し、以下のパラメータをニーズに合わせて調整できます。

項目説明推奨値
Connect TimeoutEMQXがMongoDBへの接続確立を試みる際のタイムアウト時間。30秒
Socket TimeoutMongoDBとのソケット通信でデータ送受信時にEMQXが待機する最大時間。30秒
Max Overflow Workers既存のワーカーがすべて使用中の場合に追加で生成可能なワーカー数。負荷急増時にMongoDBへの同時接続数を増やすための設定。0
Wait Queue TimeoutMongoDB接続が利用可能になるまでワーカーがアイドル状態で待機できる最大時間。10秒
Heartbeat PeriodドライバーがMongoDBの状態をチェックする間隔。MongoDBの稼働状況を監視するためのハートビート信号の送信頻度を制御。200秒
Minimum Heartbeat Periodハートビート間の最短間隔を設定し、MongoDB状態チェックの過剰な頻度を防止。EMQXとMongoDB間の効率的な通信を確保。200秒

さらに詳しく ​

以下のリンクから詳細情報をご覧いただけます：

ブログ：

MQTTとMongoDB：IoTデータ管理のためのシームレスなシナジーの構築

レポート：

MQTTパフォーマンスベンチマークテスト：EMQX-MongoDB統合

動画：

https://www.youtube.com/watch?v=c2M-rlkkT5o