EMQX Enterprise

日本語

简体中文
English

日本語

简体中文
English

Appearance

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

ルールの作成 ​

このページでは、EMQXダッシュボードを使用してデータ処理のルールを作成し、ルールにアクションを追加する方法について説明します。また、ルールのテスト方法や作成後のルールの確認方法も紹介します。

本ページのデモでは、リパブリッシュアクションを例に取り、トピック t/# で受信したメッセージを処理し、トピック a/1 にリパブリッシュするルールの作成方法を説明します。ただし、「コンソールへの結果出力」や「Sinksによる転送」もアクションの追加で言及しています。

ルールSQLの定義 ​

EMQXダッシュボードにログインし、左側のナビゲーションメニューから Integration -> Rules をクリックします。

Rules ページで Create ボタンをクリックすると、Create Rule ページに遷移します。ここでルールのデータソースを定義し、フィルタリングされたメッセージに対して実行するアクションを決定します。

ルール名を入力し、将来の管理を容易にするためにメモを追加します。SQL Editor では、ビジネスニーズに合ったデータソースを追加するためにSQL文をカスタマイズできます。本チュートリアルでは、デフォルト設定のままにしておきます。これは、"t/#" パターンにマッチするトピック（例：t/a、t/a/b、t/a/b/cなど）に属するすべてのメッセージを選択して返します。

TIP

本チュートリアルではメッセージのペイロードがJSON形式であることを想定しています。ペイロードが他の形式の場合は、スキーマレジストリを使ってデータ型を変換できます。

EMQXには豊富なSQL文のサンプルが組み込まれており、開始を支援します。SQL Editor の下にある SQL Examples ボタンをクリックして参照できます。SQLの構文や使い方の詳細はSQL Syntaxを参照してください。

image-20230417211146211

SQLジェネレーター ​

EMQX 5.10.0以降、SQL EditorはAI搭載のSQLジェネレーターを使って自然言語からルールSQLを生成できます。この機能により、自然言語で意図を記述すると、システムが適切なSQL文を自動生成します。

SQLジェネレーターはデフォルトで有効です。ダッシュボード右上の Settings メニューのトグルスイッチで無効化できます。

利用手順は以下の通りです：

  1. Create Rule ページの SQL Editor セクションに移動します。

  2. エディター下の SQL Generator ボタンをクリックして Generate SQL with AI ダイアログを開き、以下の項目を指定します：

    • Task Description（必須）：SQLに実行させたい内容を自然言語で記述します。

      例： 「MQTTメッセージのメタデータから clientid を抽出し、ペイロードから device_id と temperature を抽出する。トピックが sensors/temperature で温度が30を超えるメッセージのみ対象とする。」

    • Related Topics（任意）：sensors/temperature のようなトピックフィルターを指定します。

    • Input Example (JSON)（任意だが推奨）：AIがデータ構造を理解しやすくするためのサンプルMQTTメッセージペイロードを提供します。 例：

      json
      {
  "device_id": "sensor001",
  "temperature": 32.5,
  "unit": "C"
}

    • Output Example (JSON)（任意）：期待される結果のフォーマットを指定します。 例：

      json
      {
  "clientid": "client_a1b2c3",
  "device_id": "sensor001",
  "temperature": 32.5
}

      TIP

      入力・出力例を含めると生成されるSQLの精度が向上します。

  3. Generate をクリックして生成されたSQLをプレビューします。

  4. プレビュー画面では：

    • 生成されたSQLを確認し手動で編集可能です。
    • Apply SQL をクリックしてSQL Editorに挿入できます。
    • Back to Form をクリックして入力を修正し再生成も可能です。

  5. SQL Editorに挿入すると、SQL Editorに自動的に表示され、確認・編集できます。

出力例 ​

上記のタスクと入力例を使うと、生成されるSQLは次のようになる可能性があります：

sql
SELECT
  clientid,
  payload.device_id AS device_id,
  payload.temperature AS temperature
FROM
  "sensors/temperature"
WHERE
  payload.temperature > 30

このルールは、トピック sensors/temperature のメッセージから clientid、device_id、temperature フィールドを抽出し、温度が30を超える場合にのみ適用されます。

SQLジェネレーターを使うべき場合 ​

SQLジェネレーターは特に以下の場合に有用です：

  • EMQXのSQL構文に不慣れな場合
  • ルールのプロトタイプを素早く作成したい場合
  • 構造化されたJSONペイロードを扱う場合

より詳細なカスタマイズや構文オプションはRule SQL Syntaxドキュメントを参照してください。

SQL文のテスト ​

SQL文はシミュレーションデータを使って実行テストできます。アクションを追加しルールを作成する前に、SQLの実行結果が期待通りか検証可能です。これは任意のステップですが、EMQXルールに不慣れな場合は推奨します。ルール全体の実行テストはルールのテストを参照してください。

SQL文をテストする手順は以下の通りです：

  1. Create Rule ページで Try It Out のトグルスイッチをオンにしてSQL文のテストを有効化します。

  2. SQLに合致する Data Source を選択し、ルールで指定したソース（FROM句）と一致していることを確認します。

  3. テストデータを入力します。データソースを選択すると、EMQXは Client ID、Username、Topic、QoS、Payload など全てのシミュレーションデータフィールドにデフォルト値を提供します。必要に応じて適切な値に修正してください。

  4. Run Test ボタンをクリックしてテストを実行します。正常なら Test Passed のプロンプトが表示されます。

test-sql

SQLの処理結果はJSON形式で Output Result セクションに表示されます。SQL処理結果のすべてのフィールドは、後続のアクション（組み込みアクションやSink）で ${key} の形式で参照可能です。フィールドの詳細はSQLデータソースとフィールドを参照してください。

本デモはペイロードがJSON形式であることを想定しています。実際の運用では、スキーマレジストリを使って他形式のメッセージも扱えます。

次に、Create Rule ページ右側の Add Action ボタンをクリックして、ルールにさまざまなタイプのアクションを追加できます。

アクションの追加 ​

Create Rule ページで右側の Add Action ボタンをクリックすると、Add Action ページが表示されます。Action のドロップダウンリストから、リパブリッシュ、コンソール出力、データブリッジによる転送の3種類のアクションを選択できます。

add_action

リパブリッシュアクションの追加 ​

このセクションでは、トピック t/# で受信した元のメッセージを別のトピック a/1 にリパブリッシュするアクションの追加方法を示します。

Add Action ページで、Type of Action のドロップダウンメニューから Republish を選択し、以下の設定を行ってから Add ボタンをクリックして確定します：

  • Topic：ターゲットトピックを設定します。例では "a/1"。

  • QoS：リパブリッシュするメッセージのQoSを設定します。例では "0"。

  • Retain：このメッセージをリテインメッセージとして転送するかどうかを設定します。本チュートリアルではデフォルトの false のままにします。

  • Payload："${payload}" と入力し、リパブリッシュメッセージのペイロードが元のメッセージと同一であることを示します。

  • MQTT 5.0 メッセージプロパティ：トグルスイッチをクリックしてユーザープロパティやMQTTプロパティを必要に応じて設定します。これにより、リパブリッシュメッセージに豊富なメタデータを付加できます。

    • Payload Format Indicator：メッセージのペイロードが特定の形式かどうかを示す値を入力します。false の場合は未判別のバイト列、true の場合はUTF-8エンコードされた文字データとみなされます。これによりMQTTクライアントやサーバーがメッセージ内容を効率的に解析できます。

    • Message Expiry Interval：メッセージの有効期限（秒単位）を指定します。指定時間を過ぎるとメッセージは無効とみなされます。

    • Content Type：リパブリッシュメッセージのペイロードの種類や形式（MIMEタイプ）を指定します。例：text/plain（テキストファイル）、audio/aac（音声ファイル）、application/json（JSON形式のアプリケーションメッセージ）など。

    • Response Topic：レスポンスメッセージをパブリッシュするMQTTトピックを指定します。例：response/my_device。

    • Correlation Data：レスポンスメッセージと元のリクエストメッセージを関連付けるための一意の識別子やデータを入力します。例：リクエストIDやトランザクションIDなど。

  • Direct Dispatch：トグルスイッチで直接ディスパッチを有効または無効にします。有効にすると、メッセージはサブスクライバーに直接配信され、追加のルール発動や同一ルールの再帰的な起動を防ぎます。

注意

EMQX 6.1.5以降、ルールがネームスペースに属し、かつ rule_engine.limit_selects_in_namespace が有効（デフォルト）な場合、EMQXはリパブリッシュアクションの出力トピックをルールのネームスペース内に制限します。トピックテンプレートのレンダリング後、レンダリング結果が <namespace>/ で始まらない場合は <namespace>/ を先頭に付加します。すでに <namespace>/ で始まるトピックは変更されません。この動作は mqtt.namespace_as_mountpoint に依存しません。グローバルルールや rule_engine.limit_selects_in_namespace = false のデプロイメントは、レンダリングされたトピックにネームスペースを付加せずにパブリッシュを続けます。

詳細はルールネームスペースの分離を参照してください。

Create Rule ページ下部の Create ボタンをクリックしてルール作成を完了します。このルールは Rule ページに新しいエントリとして追加されます。

TIP

リパブリッシュアクションは元のメッセージの配信を妨げません。例えば、ルールによりトピック "t/1" のメッセージがトピック "a/1" にリパブリッシュされても、元の "t/1" メッセージは "t/1" をサブスクライブしているクライアントに引き続き配信されます。

コンソール出力アクションの追加 ​

TIP

コンソール出力アクションはデバッグ目的でのみ使用してください。本番環境で使用するとパフォーマンス問題を引き起こす可能性があります。

コンソール出力アクションはルールの出力結果を確認するために使います。結果メッセージはコンソールやログファイルに出力されます。

  • EMQXが console または foreground モードで起動された場合（Docker環境ではデフォルトで foreground）、出力はコンソールに向けられます。
  • systemd経由で起動された場合は、出力はジャーナルシステムにキャプチャされ、journalctl コマンドで確認可能です。

出力は以下の形式になります：

bash
[rule action] rule_id1
    Action Data: #{key1 => val1}
    Envs: #{key1 => val1, key2 => val2}

ここで

  • [rule action] はリパブリッシュアクションが発動したルールIDです。
  • Action Data はルールの出力結果で、アクション実行時に渡されるデータやパラメータ（リパブリッシュアクション設定時のペイロード部分）を示します。
  • Envs はリパブリッシュ時に設定される環境変数で、データソースやこのアクション実行に関連する内部情報が含まれます。

Sinksによる転送アクションの追加 ​

処理結果をSinksを使って転送するアクションも追加可能です。ダッシュボードの Type of Action ドロップダウンリストから対象のSinkを選択するだけです。EMQXの各Sinkの詳細はデータ統合を参照してください。

ルールのテスト ​

ルールエンジンはルールテスト機能を提供しており、シミュレーションデータや実際のクライアントデータを使ってルールをトリガーし、ルールSQLを実行し、ルールに追加されたすべてのアクションを実行して各ステップの結果を取得できます。

ルールをテストすることで、ルールが期待通りに動作するか検証でき、問題の早期発見と解決が可能です。これにより開発効率が向上し、本番環境での失敗を防止できます。

テスト手順 ​

  1. Try It Out スイッチをオンにし、テスト対象として Rule を選択します。テスト開始前にルールを保存しておく必要があります。

  2. Start Test ボタンをクリックしてテストを開始します。ブラウザは現在のルールがトリガーされるのを待ち、テスト結果を生成します。

  3. ルールをトリガーしてテストします。以下の2つの方法がサポートされています：

    • シミュレーションデータを使う：Input Simulated Data ボタンをクリックし、ポップアップでSQLに合致する Data Source を選択し、ルールの指定ソース（FROM句）と一致していることを確認します。EMQXはすべてのフィールド（Client ID、Username、Topic、QoS、Payload など）にデフォルト値を提供します。必要に応じて修正し、Submit Test ボタンをクリックして一度だけルールをトリガーします。

    • 実際のデバイスデータを使う：現在のページを開いたままにし、実際のクライアントまたはMQTTクライアントツールでEMQXに接続し、対応するイベントをトリガーしてテストします。

  4. テスト結果を確認します。ルールがトリガーされると、ダッシュボードに実行結果が出力され、各ステップの詳細な実行結果が表示されます。

テスト例 ​

MQTTX を使ってリパブリッシュアクションのルールをテストできます。クライアントを1つ作成し、そのクライアントでトピック a/1 をサブスクライブし、トピック t/1 にメッセージを送信します。ダイアログボックスに、このメッセージがトピック a/1 にリパブリッシュされていることが表示されます。

MQTTXクライアントツールとEMQXの接続方法の詳細は MQTTX - Get Started を参照してください。

image

対応して、ダッシュボードのテスト画面にはルール全体の実行結果が表示され、以下の内容が含まれます：

  • 左側はルールの実行記録です。ルールがトリガーされるたびに記録が生成され、クリックすると対応するメッセージやイベントの詳細に切り替わります。
  • 右側は選択したルールのアクション記録リストです。クリックするとアクションの実行結果やログを展開して確認できます。

ルールSQLやアクションの実行に失敗した場合、該当ルールの記録全体が失敗としてマークされます。記録を選択すると対応するアクションのエラー情報を確認でき、トラブルシューティングに役立ちます。

test-rules

上記例では、ルールは4回トリガーされ、そのうち3回は完全に成功しています。4回目は HTTP Server アクションの実行失敗により失敗し、エラー理由は302ステータスコードのレスポンスでした。

ルールテストの詳細な使い方はブログ記事 Enhancing Data Integration Stability: A Guide on EMQX Platform E2E Rule Testing を参照してください。

ルールの確認 ​

Rules ページでは、作成したすべてのルールの一覧を包括的に確認できます。

一覧の各エントリには、ルールID、関連するソース、有効状態、アクション数などの基本情報が表示されます。ソースにマウスオーバーすると対応するSQL文の詳細が表示されます。ルールの設定を変更するには、Actions 列の Settings をクリックします。More ボタンからはルールの複製や削除も可能です。

view_rules

また、Flowデザイナーで Integration -> Flow Designer に移動してルールを確認することもできます。Rules ページで作成したルールとFlowデザイナーで作成したルールは完全に相互運用可能です。

ルールの統計情報やアクション実行情報を確認するには、Rules ページのルールIDまたは Flows ページのルール名をクリックします。

view_rules_flows

TIP

ルールアクションを更新したりデータソースを再定義した場合、以下のページに表示される統計情報はリセットされて再集計が始まります。

Rule Statistics

ルールの検索 ​

ルールが多数ある場合はフィルターを使って検索を絞り込み、表示したいルールだけを表示できます。ルールID、受信メッセージのトピックやワイルドカード、有効状態、ルールメモ、ルールに関連付けられたアクションやソースでフィルター可能です。

search_rules

アクション（Sink）とソースの確認 ​

Rule ページの Actions (Sink) タブと Sources タブには、作成済みのすべてのアクション（Sink）とソースが表示されます。これらのタブには名前、接続状態、関連ルール、有効状態、作成日時、最終更新日時などの重要な情報が含まれます。列名横の矢印をクリックするとソートできます。

Enable 列のトグルスイッチをクリックすると、Sinkやソースの有効・無効を切り替えられます。Associated Rules 列の View Rules をクリックすると、そのSinkやソースを含むルールの一覧が表示され、データ統合設定の管理が容易になります。

Action 列からはSinkやソースの再接続や設定変更が可能です。More をクリックすると、Sinkやソースの削除や、それを使った新規ルールの作成ができます。

Sinkやソースが多数ある場合はフィルターを使って検索を絞り込み、表示したいエントリだけを表示できます。名前、状態、有効状態でフィルター可能です。

view_sink_source

Sinkやソースの統計情報やレート指標を確認するには、名前をクリックします。

action_statistics