Merge pull request #2679 from emqx/20240927-r58-external-registries

Meggielqk · web-flow · commit 3bbccd35de11 · 2024-10-14T16:55:01.000+08:00
feat: external schema registry
diff --git a/en_US/data-integration/schema-registry.md b/en_US/data-integration/schema-registry.md
@@ -11,15 +11,15 @@ Because of the variety of IoT device terminals and the different coding formats
 The Schema Registry manages the Schema used for coding and decoding, processes the encoding or decoding requests, and returns the results. The Schema Registry in collaboration with the rule engine can be adapted for device access and rule design in
 various scenarios.
 
-EMQX Schema Registry currently supports codecs in below formats:
+EMQX Schema Registry currently supports codecs in the below formats:
 
 - [Avro](https://avro.apache.org)
 - [Protobuf](https://developers.google.com/protocol-buffers/)
 - [JSON Schema](https://json-schema.org/)
 
 Avro and Protobuf are Schema-dependent data formats. The encoded data is binary and the decoded data is in [Map format](#rule-engine-internal-data-format-map). The decoded data can be used directly by the rule engine and other plugins. Schema Registry maintains Schema text for built-in encoding formats such as Avro and Protobuf.
 
-JSON schema can be used to validate if input JSON object is following the schema definetions or if the JSON object output from the rule engine is valid before producing the data to downstream.
+JSON schema can be used to validate if the input JSON object is following the schema definitions or if the JSON object output from the rule engine is valid before producing the data to downstream.
 
 The diagram below shows an example of a Schema Registry application. Multiple devices report data in different formats, which are decoded by Schema Registry into a uniform internal format and then forwarded to the backend application.
 
@@ -105,3 +105,134 @@ The SQL statement above will match an MQTT message with the content of the paylo
 
 **Note:** The `AS` clause is required to assign the decoded data to a key so that subsequent operations can be performed on it later.
 
+
+## External Schema Registry
+
+Starting with version 5.8.1, EMQX supports configuring an external Confluent Schema Registry (CSR). This feature allows users to dynamically retrieve schemas from external registries during rule processing, enabling efficient message encoding and decoding.
+
+### Create External Schema Registry in Dashboard
+
+You can configure an external schema registry directly through the EMQX Dashboard, making it easy to manage your schema integration.
+
+Go to **Integration** -> **Schema** on EMQX Dashboard. Select the **External** tab on the Schema page.
+
+Click the **Create** button at the upper right corner. Configure the following fields:
+
+- **Name**: Enter an external schema registry name that will be used in the encoding and decoding functions.
+- **Type**: Select the type of external schema registry. Currently, only `Confluent` is supported.
+- **URL**: Enter the endpoint of your Confluent Schema Registry.
+- **Authentication**: If you select `Basic auth`, enter the authentication credentials (username and password) for accessing the external registry.
+
+Click **Create** after you complete the settings.
+
+### Configure External Schema Registry via Configuration File
+
+You can configure an external Confluent Schema Registry via the EMQX configuration file. Here’s an example of how to set it up:
+
+
+```hcl
+schema_registry {
+  external {
+    my_external_registry {
+      type = confluent
+      url = "https://confluent.registry.url:8081"
+      auth {
+        username = "myuser"
+        password = "secret"
+      }
+    }
+  }
+}
+```
+
+In this example:
+
+- `my_external_registry` is the name assigned to the external schema registry.
+- `type = confluent` specifies the type of external registry.
+- `url` is the endpoint of your Confluent Schema Registry.
+- `auth` contains the authentication credentials (username and password) for accessing the external registry.
+
+### Use External Schema Registry in Rule Engine
+
+Once an external registry is configured, you can use several functions in the EMQX rule engine to encode and decode payloads using the schemas stored in the external registry.
+
+The following functions utilize a configured external 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)
+```
+
+In all function usage examples below, the following example values and variable names are used:
+
+- `my_external_registry` is the name you assigned to the external registry in EMQX.
+- `my_schema_id` is the schema ID registered in the CSR (always an integer in CSR).
+- `my_local_avro_schema` is the name of a locally configured Avro schema in EMQX.
+- `my_subject` is the subject name defined in the CSR.
+
+#### Function Usage Examples
+
+##### `avro_encode`
+
+`avro_encode` encodes a payload using the schema ID from the external registry. The schema is retrieved dynamically at runtime and cached for subsequent runs. In Confluent Schema Registry, schema IDs are integers.
+
+::: tip Note
+
+When encoding, the payload must be in the internal data format of the rule engine, which is a decoded map. This is why `json_decode` is used in the example.
+
+:::
+
+Example:
+
+```sql
+select
+  -- 123 is the schema ID that is registered in CSR
+  avro_encode('my_external_registry', json_decode(payload), 123) as encoded
+from 't'
+```
+
+##### `avro_decode`
+
+This function decodes an Avro payload based on the specified schema ID from the external registry. The schema is dynamically fetched during runtime and cached for subsequent operations.
+
+Example:
+
+```sql
+select
+  -- 123 is the schema ID that is registered in CSR
+  avro_decode('my_external_registry', payload, 123) as decoded
+from 't'
+```
+
+##### `schema_encode_and_tag` 
+
+This function uses a locally registered Avro schema, an external CSR schema name, and a subject to encode a payload (already in internal map format), and to tag the resulting payload with a schema ID.  The schema ID comes from registering the local schema to CSR.
+
+Example:
+
+```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` 
+
+This function uses a CSR name to decode a payload, assuming it is tagged with the schema ID retrieved from CSR.
+
+```sql
+select
+  schema_decode_tagged(
+    'my_external_registry',
+    payload
+  ) as decoded
+from 't'
+```
+
diff --git a/zh_CN/data-integration/schema-registry.md b/zh_CN/data-integration/schema-registry.md
@@ -58,7 +58,7 @@ schema_check(SchemaName, Map | Bytes) -> Boolean
 
 ## 编解码 + 规则引擎
 
-EMQX 的消息处理层面可分为消息路由(Messaging)、规则引擎(Rule Engine)、数据格式转换(Data Conversion) 三个部分。
+EMQX 的消息处理层面可分为消息路由 (Messaging)、规则引擎 (Rule Engine)、数据格式转换 (Data Conversion) 三个部分。
 
 EMQX 的 PUB/SUB 系统将消息路由到指定的主题。规则引擎可以灵活地配置数据的业务规则，按规则匹配消息，然后指定相应动作。数据格式转换发生在规则匹配的过程之前，先将数据转换为可参与规则匹配的 Map 格式，然后进行匹配。
 
@@ -100,3 +100,132 @@ SELECT json_decode(payload) AS p FROM "t/#" WHERE p.x = p.y
 ```
 
 **注意:** `AS` 子句是必须的，将解码之后的数据赋值给某个Key，后面才能对其进行后续操作。
+
+## 外部 Schema Registry
+
+从 EMQX 版本 5.8.1 开始，支持在 EMQX 中配置外部 Confluent Schema Registry (CSR)。该功能允许用户在规则处理时动态获取外部 Schema Registry 中的 Schema，从而实现高效的消息编码和解码。
+
+### 在 Dashboard 中创建外部 Schema Registry
+
+您可以直接通过 EMQX Dashboard 配置外部 Schema Registry，方便地管理 Schema 集成。
+
+进入 EMQX Dashboard 的 **集成** -> **Schema** 页面。在 Schema 页面中选择 **外部 Schema** 选项卡。
+
+点击右上角的**创建**按钮，并配置以下字段：
+
+- **名称**：输入外部 Schema Registry 的名称，该名称将在编码和解码函数中使用。
+- **类型**：选择外部 Schema Registry 的类型。目前仅支持 `Confluent`。
+- **URL**：输入您的 Confluent Schema Registry 的端点地址。
+- **认证**：如果选择 `基础认证`，请输入访问外部 Schema Registry 所需的认证信息（用户名和密码）。
+
+完成设置后，点击**创建**按钮。
+
+### 通过配置文件配置外部 Schema Registry
+
+您也可以通过 EMQX 配置文件配置外部 Confluent Schema Registry。以下是配置示例：
+
+```hcl
+schema_registry {
+  external {
+    my_external_registry {
+      type = confluent
+      url = "https://confluent.registry.url:8081"
+      auth {
+        username = "myuser"
+        password = "secret"
+      }
+    }
+  }
+}
+```
+
+在此示例中：
+
+- `my_external_registry` 是分配给外部 Schema Registry 的名称。
+- `type = confluent` 指定外部 Schema Registry 的类型。
+- `url` 是 Confluent Schema Registry 的端点地址。
+- `auth` 包含访问外部 Schema Registry 所需的认证信息（用户名和密码）。
+
+### 在规则引擎中使用外部 Schema Registry
+
+配置外部 Schema Registry 后，您可以在 EMQX 规则引擎中使用多个函数，利用外部 Schema Registry 中存储的 Schema 对 payload 进行编码和解码。
+
+配置的外部 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 中为外部 Schema Registry 指定的名称。
+- `my_schema_id` 是注册在 CSR 中的 Schema ID（在 CSR 中始终是整数）。
+- `my_local_avro_schema` 是在 EMQX 中配置的本地 Avro Schema 名称。
+- `my_subject` 是在 CSR 中定义的主题名称。
+
+#### 函数使用示例
+
+##### `avro_encode`
+
+`avro_encode` 使用外部 Schema Registry 中的 Schema ID 对 payload 进行编码。Schema 会在运行时动态获取，并缓存以供后续使用。在 Confluent Schema Registry 中，Schema ID 是整数。
+
+::: tip 提示
+
+编码时，payload 必须是规则引擎的内部数据格式，即已解码的 Map。因此在示例中使用了 `json_decode`。
+
+:::
+
+示例：
+
+```sql
+select
+  -- 123 是在 CSR 中注册的 Schema ID
+  avro_encode('my_external_registry', json_decode(payload), 123) as encoded
+from 't'
+```
+
+##### `avro_decode`
+
+该函数根据外部 Schema Registry 中的 Schema ID 对 Avro payload 进行解码。Schema 会在运行时动态获取，并缓存以供后续操作。
+
+示例：
+
+```sql
+select
+  -- 123 是在 CSR 中注册的 Schema ID
+  avro_decode('my_external_registry', payload, 123) as decoded
+from 't'
+```
+
+##### `schema_encode_and_tag`
+
+此函数使用本地注册的 Avro Schema、外部 CSR 的 Schema 名称和主题对 payload 进行编码，并将编码后的 payload（已为内部 Map 格式）标记为带有 Schema ID。Schema ID 是通过将本地 Schema 注册到 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 名称对 payload 进行解码，假设该 payload 带有从 CSR 获取的 Schema ID。
+
+```sql
+select
+  schema_decode_tagged(
+    'my_external_registry',
+    payload
+  ) as decoded
+from 't'
+```
