EMQX Enterprise

日本語

简体中文
English

日本語

简体中文
English

Appearance

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

スキーマレジストリ ​

EMQX スキーマレジストリは、MQTT メッセージのペイロードのエンコード、デコード、および検証のためのスキーマを定義・管理する機能を提供します。ルールはスキーマレジストリの関数を呼び出して、Avro や Protobuf といったバイナリペイロードをルールエンジンが処理可能なデータにデコードしたり、処理済みデータを下流システム向けにエンコードしたり、JSON データを JSON スキーマに対して検証したりできます。

デバイスと下流アプリケーションが異なるフォーマットでデータをやり取りする場合にスキーマレジストリを使用します。スキーマ定義やカスタムコーデック設定を一元管理することで、各ルールやアプリケーションで個別に変換ロジックを実装することなく、一貫したフォーマットでメッセージを処理できます。

以下の図はスキーマレジストリの利用例です。複数のデバイスが異なるフォーマットでデータを報告し、スキーマレジストリで統一された内部フォーマットにデコードされてからバックエンドアプリケーションに転送されます。

スキーマレジストリ

対応スキーマタイプ ​

EMQX スキーマレジストリは以下の内部スキーマタイプをサポートしています。

スキーマタイプ説明例
AvroMap フォーマット から Avro バイナリデータへのエンコード、および Avro バイナリデータから Map フォーマットへのデコードを行います。スキーマレジストリの例 - Avro
ProtobufMap フォーマットから Protobuf バイナリデータへのエンコード、および Protobuf バイナリデータから Map フォーマットへのデコードを行います。スキーマレジストリの例 - Protobuf
JSON Schema入力された JSON データやルールエンジンが生成した JSON データが JSON スキーマに準拠しているかを検証します。スキーマレジストリの例 - JSON Schema
外部 HTTP サーバーカスタムコーデックロジックを実装した HTTP サービスにペイロードのエンコード・デコードを委譲します。スキーマレジストリの例 - 外部 HTTP サーバー

外部 HTTP サーバーと外部スキーマレジストリは別の統合機能です。外部 HTTP サーバーは内部スキーマタイプの一つで、エンコード・デコード処理をカスタム HTTP サービスに委譲します。一方、外部スキーマレジストリは別途設定され、ルール処理時に設定された Confluent スキーマレジストリから Avro スキーマを取得します。詳細は外部スキーマレジストリをご参照ください。

JSON Schema サポート ​

EMQX 6.0.4 以降、スキーマレジストリは JSON Schema draft-03、draft-04、draft-06、draft 2019-09、draft 2020-12 をサポートしています。EMQX は $schema フィールドの値に基づいて JSON Schema のバージョンを選択し、$schema が省略された場合は draft-06 を使用します。

完全な例や各ドラフトの制限については、スキーマレジストリの例 - JSON Schemaをご覧ください。

アーキテクチャ設計 ​

EMQX はパブリッシュされたメッセージのエンコード、デコード、およびスキーマ仕様への準拠検証にスキーマを利用できます。Avro や Protobuf などの組み込みエンコードフォーマットのスキーマテキストを管理しています。

スキーマ API はスキーマ名による追加、照会、削除操作を提供するため、エンコード・デコード時にはスキーマ名の指定が必要です。

architecture

一般的なユースケースとして、ルールエンジンがスキーマレジストリのエンコード・デコードインターフェースを呼び出し、その結果のデータを後続のアクションの入力として利用します。

エンコード呼び出しの例：

erlang
schema_encode(SchemaName, Map) -> Bytes

デコード呼び出しの例：

erlang
schema_decode(SchemaName, Bytes) -> Map

JSON エンコードされた MQTT メッセージのデータをエンコードする場合は、スキーマ関数でエンコードする前に json_decode 関数を使って Map 内部フォーマットにデコードする必要があります。例えば：

erlang
schema_encode(SchemaName, json_decode(JSONData)) -> Bytes

JSON データが JSON スキーマに準拠しているかをエンコード前またはデコード後に検証する場合は、以下のスキーマ検証例を使用します。

erlang
schema_check(SchemaName, Map | Bytes) -> Boolean

スキーマレジストリとルールエンジン ​

EMQX のメッセージ処理層は、メッセージング、ルールエンジン、データ変換の3つに分けられます。

EMQX の PUB/SUB システムはメッセージを指定されたトピックにルーティングします。ルールエンジンはデータに対するビジネスルールを柔軟に設定でき、メッセージをルールにマッチさせて対応するアクションを指定します。データ形式の変換はルールマッチング処理の前に行われ、マッチングに参加可能な Map フォーマットに変換されてからマッチングが行われます。

スキーマとルールエンジン

ルールエンジン内部データフォーマット（Map） ​

ルールエンジン内部で使用されるデータフォーマットは Erlang の Map であるため、元のデータがバイナリやその他のフォーマットの場合は、上記の schema_decode や json_decode のようなコーデック関数で Map に変換する必要があります。JSON オブジェクトに非常に似ています。

Map は #{key => value} の形式を持つキー・バリュー型のデータ構造です。例えば、user = #{id => 1, name => "Steve"} は id が 1、name が "Steve" の user Map を定義しています。

SQL 文では . 演算子を使ってネストされた Map フィールドの抽出や追加が可能です。以下は SQL 文での Map 操作例です。

sql
SELECT user.id AS my_id

この SQL 文のフィルター結果は #{my_id => 1} となります。

JSON コーデック ​

ルールエンジンの SQL 文は JSON 形式の文字列のエンコード・デコードをサポートしています。JSON 文字列を Map フォーマットに変換する SQL 関数は json_decode() と json_encode() です。

sql
SELECT json_decode(payload) AS p FROM "t/#" WHERE p.x = p.y

上記の SQL 文は、ペイロードの内容が JSON 文字列 {"x": 1, "y": 1} でトピックが t/a の MQTT メッセージにマッチします。

json_decode(payload) as p は JSON 文字列を以下の Map データ構造にデコードし、WHERE 句で p.x や p.y としてフィールドを利用できるようにします。

erlang
#{
  p => #{
    x => 1,
    y => 1
  }
}

注意: AS 句はデコードしたデータをキーに割り当てるために必要で、後続の操作で利用可能にします。

外部スキーマレジストリ ​

EMQX 5.8.1 以降、外部 Confluent スキーマレジストリ（CSR）の設定をサポートしています。この機能により、ルール処理中に外部レジストリから動的にスキーマを取得し、効率的なメッセージのエンコード・デコードを実現します。

ダッシュボードでの外部スキーマレジストリ作成 ​

EMQX ダッシュボードから直接外部スキーマレジストリを設定でき、スキーマ統合の管理が容易です。

EMQX ダッシュボードの Smart Data Hub -> Schema Registry に移動し、スキーマページの External タブを選択します。

右上の Create ボタンをクリックし、以下の項目を設定します。

  • Name: エンコード・デコード関数で使用する外部スキーマレジストリ名を入力します。
  • Type: 外部スキーマレジストリのタイプを選択します。現在は Confluent のみ対応しています。
  • URL: Confluent スキーマレジストリのエンドポイントを入力します。
  • Authentication: Basic auth を選択した場合、外部レジストリにアクセスするための認証情報（ユーザー名とパスワード）を入力します。

設定完了後、Create をクリックします。

設定ファイルによる外部スキーマレジストリの設定 ​

EMQX の設定ファイルで外部 Confluent スキーマレジストリを設定する例は以下の通りです。

hcl
schema_registry {
  external {
    my_external_registry {
      type = confluent
      url = "https://confluent.registry.url:8081"
      auth {
        username = "myuser"
        password = "secret"
      }
    }
  }
}

この例では、

  • my_external_registry が外部スキーマレジストリに割り当てた名前です。
  • type = confluent は外部レジストリのタイプを指定しています。
  • url は Confluent スキーマレジストリのエンドポイントです。
  • auth は外部レジストリにアクセスするための認証情報（ユーザー名とパスワード）です。

ルールエンジンでの外部スキーマレジストリの利用 ​

外部レジストリを設定すると、EMQX ルールエンジンで外部レジストリに保存されたスキーマを使ってペイロードのエンコード・デコードを行う関数を利用できます。

以下の関数は設定済みの外部 CSR を利用します。

sql
avro_encode('my_external_registry', payload, my_schema_id)
avro_decode('my_external_registry', payload, my_schema_id)
schema_encode_and_tag('my_local_avro_schema', 'my_external_registry', payload, 'my_subject')
schema_decode_tagged('my_external_registry', payload)

関数利用例 ​

以下の例では、以下の値と変数名を使用しています。

  • my_external_registry は EMQX で外部レジストリに割り当てた名前です。
  • my_schema_id は CSR に登録されたスキーマ ID（CSR では常に整数）です。
  • my_local_avro_schema は EMQX にローカル設定された Avro スキーマ名です。
  • my_subject は CSR で定義されたサブジェクト名です。
avro_encode ​

avro_encode は外部レジストリのスキーマ ID を使ってペイロードをエンコードします。スキーマは実行時に動的に取得され、後続の処理でキャッシュされます。Confluent スキーマレジストリではスキーマ ID は整数です。

注意

エンコード時のペイロードはルールエンジンの内部データフォーマットであるデコード済み Map である必要があります。これが例で json_decode を使う理由です。

例：

sql
select
  -- 123 は CSR に登録されたスキーマ ID
  avro_encode('my_external_registry', json_decode(payload), 123) as encoded
from 't'
avro_decode ​

この関数は外部レジストリの指定されたスキーマ ID に基づいて Avro ペイロードをデコードします。スキーマは実行時に動的に取得され、後続の処理でキャッシュされます。

例：

sql
select
  -- 123 は CSR に登録されたスキーマ ID
  avro_decode('my_external_registry', payload, 123) as decoded
from 't'
schema_encode_and_tag ​

この関数はローカルに登録された Avro スキーマ、外部 CSR スキーマ名、サブジェクトを使って、（すでに内部 Map フォーマットの）ペイロードをエンコードし、スキーマ ID でタグ付けします。スキーマ ID はローカルスキーマを CSR に登録した際に得られます。

例：

sql
select
  schema_encode_and_tag(
    'my_local_avro_schema',
    'my_external_registry',
    json_decode(payload),
    'my_subject'
  ) as encoded
from 't'
schema_decode_tagged ​

この関数は CSR 名を使って、スキーマ ID でタグ付けされたペイロードをデコードします。

sql
select
  schema_decode_tagged(
    'my_external_registry',
    payload
  ) as decoded
from 't'