EMQX Platform

日本語

简体中文
English

日本語

简体中文
English

Appearance

お問い合わせ

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

Apache Cassandra は、大規模データセットの処理と高スループットアプリケーションの構築に適した、オープンソースの分散型NoSQLデータベース管理システムです。EMQX PlatformとApache Cassandraの連携により、Cassandraデータベースへのメッセージやイベントの保存が可能となり、時系列データの保存、デバイス登録・管理、リアルタイムデータ分析などの機能を実現できます。

本ページでは、EMQX PlatformとCassandra間のデータ統合について、実践的な手順を交えて包括的に解説します。

TIP

現在の実装はCassandra v3.xのみ対応しており、v4.xには未対応です。

動作概要 ​

Cassandraデータ統合はEMQX Platformの標準機能であり、EMQX Platformのデバイス接続およびメッセージ送受信機能とCassandraの強力なデータ保存機能を組み合わせています。組み込みのルールエンジンコンポーネントにより、EMQX PlatformからCassandraへのデータ取り込みを簡素化し、複雑なコーディングを不要にします。

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

EMQX Platform Integration Cassandra

MQTTデータのCassandraへの取り込みは以下の流れで行われます。

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

特長と利点 ​

Cassandraとのデータ統合は、効率的なデータ送信、保存、活用を実現するための多彩な特長と利点を提供します。

  • 大規模時系列データ保存：EMQX Platformは大量のデバイス接続とメッセージ送信を処理可能です。Cassandraの高いスケーラビリティと分散ストレージ機能を活用し、大規模データセット（時系列データを含む）の保存・管理が可能で、時間範囲に基づくクエリや集計操作をサポートします。
  • リアルタイムデータストリーミング：EMQX Platformはリアルタイムデータストリーム処理に最適化されており、ソースシステムからCassandraへの効率的かつ信頼性の高いデータ送信を保証します。即時の洞察やアクションが必要なユースケースに適しています。
  • 高可用性の保証：EMQXとCassandraの両方がクラスタリング機能を提供します。組み合わせて使用することで、デバイス接続やデータを複数サーバーに分散可能です。ノード障害時には他の利用可能なノードへ自動的に切り替わり、高いスケーラビリティとフォールトトレランスを実現します。
  • 柔軟なデータ変換：EMQX Platformの強力なSQLベースのルールエンジンにより、Cassandraへの保存前にデータを前処理できます。フィルタリング、ルーティング、集計、強化など多様な変換機構をサポートし、ニーズに応じたデータ整形が可能です。
  • 柔軟なデータモデル：Cassandraはカラムベースのデータモデルを採用し、柔軟なスキーマ設計およびカラムの動的追加をサポートします。構造化されたデバイスイベントやメッセージデータの保存・管理に適しており、多様なMQTTメッセージデータを容易に格納できます。

はじめる前に ​

このセクションでは、TimescaleDBデータブリッジ作成の準備として、Cassandraサーバーのインストールやキー スペース・テーブルの作成方法について説明します。

前提条件 ​

ネットワーク設定 ​

EMQX Platform上でデプロイメント（EMQXクラスター）を作成し、ネットワークを構成する必要があります。

  • 専用デプロイメントユーザーの場合：まず、VPCピアリング接続を作成してください。ピアリング接続が確立された後、内部ネットワークのIPを介してプラットフォームコンソールにログインし、対象のコネクターにアクセスできます。あるいは、NATゲートウェイを設定して、パブリックIP経由で対象のコネクターにアクセスすることも可能です。
  • BYOC（Bring Your Own Cloud）デプロイメントユーザーの場合：BYOCがデプロイされているVPCと対象コネクターが存在するVPC間でピアリング接続を確立してください。ピアリング接続作成後、内部ネットワークのIPを介して対象コネクターにアクセスできます。パブリックIPアドレス経由でリソースにアクセスする必要がある場合は、パブリッククラウドコンソールでBYOCがデプロイされているVPCに対してNATゲートウェイを構成してください。

Cassandraサーバーのインストール ​

以下のコマンドでDockerを使い、シンプルなCassandraサービスを起動します。

bash
docker run --name cassa --rm -p 9042:9042 cassandra:3.11.14

キースペースとテーブルの作成 ​

データブリッジ作成前に、キースペースとテーブルを作成する必要があります。

  1. mqttという名前のキースペースを作成します。
bash
docker exec -it cassa cqlsh "-e CREATE KEYSPACE mqtt WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1}"
  1. Cassandraにtemp_humテーブルを作成します。
bash
docker exec -it cassa cqlsh "-e \
    CREATE TABLE mqtt.temp_hum( \
        msgid text, \
        temp text, \
        hum text, \
        arrived timestamp, \
        PRIMARY KEY(msgid));"

Cassandraコネクターの作成 ​

データ統合ルールを作成する前に、CassandraサーバーにアクセスするためのCassandraコネクターを作成する必要があります。

  1. デプロイメントにアクセスし、左側ナビゲーションメニューからデータ統合をクリックします。

  2. 初めてコネクターを作成する場合は、データ永続化カテゴリの中からCassandraを選択します。すでにコネクターを作成済みの場合は、新規コネクターを選択し、続いてデータ永続化カテゴリのCassandraを選択します。

  3. 接続情報を入力します。

    • Servers：サーバーのIPアドレスとポート
    • Keyspace：mqtt
    • その他はデフォルトのままにします。
    • 暗号化接続を確立したい場合は、TLSを有効にするのトグルスイッチをオンにします。

  4. 詳細設定（任意）

  5. テストボタンをクリックし、Cassandraサービスにアクセス可能であれば成功メッセージが表示されます。

  6. 新規作成ボタンをクリックして作成を完了します。

ルールの作成 ​

次に、書き込み対象のデータを指定し、処理済みデータをCassandraに転送するアクションをルールに追加します。

  1. ルールエリアで新規ルールをクリックするか、作成したコネクターのアクション列にある新規ルールアイコンをクリックします。

  2. SQLエディターにルールのマッチングSQL文を入力します。以下の例では、メッセージ到着時刻arrived、クライアントID、temp_hum/emqxトピックのペイロードから温度と湿度を読み取ります。

    sql
      SELECT
    id as msgid,
    payload.temp as temp,
    payload.hum as hum,
    timestamp as arrived
  FROM
    "temp_hum/emqx"

    TIP

    初心者の方は、SQL例をクリックし、テストを有効にするを使ってSQLルールの学習と検証が可能です。

  3. 次へをクリックしてアクションを追加します。

  4. コネクターのドロップダウンから先ほど作成したコネクターを選択します。

  5. CQLテンプレートを設定し、msgid、temp、hum、arrivedをCassandraに保存します。このテンプレートはCassandra Query Languageで実行され、サンプルコードは以下の通りです。

    sql
      INSERT INTO temp_hum(msgid, temp, hum, arrived)
  VALUES (
    ${msgid},
    ${temp},
    ${hum},
    ${arrived}
  )

  6. 詳細設定（任意）

  7. 確定ボタンをクリックしてルール作成を完了します。

  8. 新規ルール成功のポップアップでルールに戻るをクリックし、データ統合設定の一連の流れが完了します。

ルールのテスト ​

MQTTXを使って温度・湿度データの送信をシミュレートすることを推奨しますが、他の任意のクライアントでも可能です。

  1. MQTTXでデプロイメントに接続し、以下のトピックにメッセージを送信します。

    • トピック：temp_hum/emqx

    • ペイロード：

      json
      {
  "temp": "27.5",
  "hum": "41.8"
}

  2. 以下のコマンドでメッセージがCassandraに保存されているか確認します。

    bash
     $ docker exec -it cassa cqlsh "-e SELECT * FROM mqtt.temp_hum;"

 msgid                            | arrived                         | hum  | temp
 ----------------------------------+---------------------------------+------+------
 00061488D7FBFE8F2C770000467D0011 | 2024-03-26 04:37:11.987000+0000 | 41.8 | 27.5

 (1 rows)

  3. コンソールで運用データを確認します。ルール一覧のルールIDをクリックすると、ルールの統計情報およびそのルールに紐づく全アクションの統計情報が表示されます。