# 将 MQTT 数据写入到 Tablestore

[Tablestore](https://www.aliyun.com/product/ots) 是一款可扩展的无服务器数据库，专为物联网场景优化。它提供了一种名为 IoTstore 的一站式解决方案，用于管理时序数据、结构化数据和半结构化数据。非常适用于物联网、车联网、风险控制、消息传递和推荐系统等场景。Tablestore 提供具有成本效益的高性能数据存储，支持毫秒级查询、检索，并具备灵活的数据分析能力。EMQX Cloud 与 Tablestore 云版、Tablestore OSS 和 Tablestore 企业版无缝集成，帮助实现物联网应用场景下的高效数据管理。

## 工作原理

EMQX Cloud 中的 Tablestore 数据集成将 EMQX Cloud 的实时数据捕获和传输能力与 Tablestore 的高性能数据存储和分析功能无缝结合。通过利用内置的[规则引擎](./rules.md)，该集成简化了将 EMQX Cloud 中的数据摄取并存储到 Tablestore 的过程，避免了复杂的编码工作。EMQX Cloud 通过其规则引擎和 Sink 将物联网设备数据转发到 Tablestore，实现高效存储和分析。

数据存储完成后，Tablestore 提供强大的分析工具，包括生成报告、图表和其他可视化的能力，用户可以通过 Tablestore 的可视化功能查看这些分析结果。

下图展示了 EMQX Cloud 和 Tablestore 数据集成的典型 Flow。

![通过 EMQX Cloud Flow 将 MQTT 数据写入 Tablestore](./_assets/tablestore-flow.png)

EMQX Cloud 和 Tablestore 提供了一个可扩展的物联网平台，用于高效地实时收集和分析设备数据。在此架构中，EMQX Cloud 作为物联网平台，负责设备接入、消息传输、数据路由等功能，Tablestore 作为数据存储和分析平台，负责数据存储、数据分析等功能。具体的工作流程如下：

1. **消息发布与接收**：设备通过 MQTT 协议连接成功后定期发布遥测数据。EMQX Cloud 接收到消息后将在规则引擎中进行比对。
2. **规则引擎处理消息**：通过内置的规则引擎，可以根据主题匹配处理特定来源的消息。当消息到达时，它会通过规则引擎，规则引擎会匹配对应的规则，并对消息数据进行处理，例如转换数据格式、过滤掉特定信息或使用上下文信息丰富消息。
3. **写入到 Tablestore**：规则引擎中定义的规则触发将消息写入到 Tablestore 的操作。Tablestore Sink 提供了可配置的字段，能够灵活地定义写入的数据格式，将消息中的特定字段写入到 Tablestore 的对应表和列中。

设备数据写入到 Tablestore 后，您可以对数据进行分析，例如：

- 连接到可视化工具，例如 Grafana，根据数据生成图表，展示设备数据。
- 连接业务系统，进行设备状态监控与告警。

## 特性与优势

Tablestore 数据集成具有以下特性与优势：

- **高效的数据处理能力**：EMQX Cloud 能够处理海量物联网设备连接与消息吞吐，Tablestore 在数据写入、存储和查询方面具有出色的性能表现，能够满足物联网场景下的数据处理需求，不会导致系统不堪重负。
- **消息转换**：消息可以在写入 Tablestore 之前，通过 EMQX Cloud 规则进行丰富的处理和转换。
- **可扩展性**：EMQX Cloud 与 Tablestore 都具备集群扩展能力，能够随着业务的发展进行集群水平扩展，满足业务的发展需求。
- **丰富的查询能力**：Tablestore 提供包括优化的函数、运算符和索引技术，可实现对时间戳数据的高效查询和分析，准确地从 IoT 时间序列数据中提取有价值的见解。
- **高效存储**：Tablestore 使用高压缩比的编码方式，可以大幅降低存储成本。也可以自定义不同数据的存储时间，避免不必要的数据占用存储空间。

## 准备工作

本节介绍了在 EMQX Cloud 中创建 Tablestore Sink 之前需要完成的准备工作，包括创建数据库实例、创建时序表和管理时序表。

> EMQX Cloud 与 Tablestore 的数据集成目前仅支持时序模型。因此，以下步骤将专注于介绍时序模型的相关数据集成操作。

### 前置准备

- 了解[规则](./rules.md)。
- 了解[数据集成](./introduction.md)。

### 网络设置

开始之前，您需要在 EMQX Cloud 上创建一个部署（EMQX 集群）并配置网络。

- 对于专有版和弹性专有版部署用户：请先创建 [VPC 对等连接](https://docs.emqx.com/zh/cloud/latest/deployments/vpc_peering.html)。创建完成后，可以通过内部网络地址访问目标连接器。或者开通 [NAT 网关](https://docs.emqx.com/zh/cloud/latest/vas/nat-gateway.html)，通过公网地址访问目标连接器。
- 对于 BYOC 部署用户：请在部署 BYOC 的 VPC 和目标连接器所在的 VPC 之间建立对等连接。创建完成后，可以通过内部网络 IP 访问目标连接器。如果需要通过公网地址访问资源，请在公共云控制台中为部署 BYOC 的 VPC 配置 NAT 网关。

### 创建时序表

1. 登录到 [Tablestore 控制台](https://otsnext.console.aliyun.com/)。
2. 创建时序模型数据库实例。有关创建实例的详细说明，请参考 [Tablestore 官方文档](https://help.aliyun.com/zh/tablestore/)。
3. 转到实例管理页面。
4. 在实例详情页签中，选择**时序表列表**，并点击**创建时序表**按钮。
5. 配置时序表信息，为时序表提供一个名称。本示例使用 `telemetry`。点击**确定**。

![在 Tablestore 实例中创建时序表](./_assets/tablestore-create-table.png)

### 管理时序表

要管理之前创建的时序表，请点击表格名称进入时序表管理界面。在那里，您可以根据业务需求执行以下操作：

1. 点击**数据管理**页签。
2. 点击**新增时间线**。

> 此步骤是可选的。如果时间线尚未创建，Tablestore 在数据写入时会自动创建时间线。因此，本示例未演示手动添加时间线的步骤。

![在 Tablestore 中新增时间线](./_assets/tablestore-add-timeline.png)

## 创建连接器

本节演示如何创建一个用于将 Sink 连接到 Tablestore 服务器的连接器。

1. 登录 EMQX Cloud，进入部署，点击**数据集成** > **连接器**。
2. 点击页面右上角的**新建连接器**。
3. 在连接器类型中选择 **Alibaba 表格存储**。
4. 配置以下信息：
   - 输入连接器名称，名称应由字母和数字的组合组成。示例：`my_tablestore`。
   - 输入 Tablestore 服务器连接信息：
     - **端点**：输入 Tablestore 实例的访问地址。您可以在 Tablestore 控制台的实例详情页面找到该地址。根据网络连接方式填写公网地址或 VPC 地址。
     - **实例名称**：要连接的 Tablestore 实例名称。
     - **访问密钥 ID**：用于与 Tablestore 认证的 AccessKey ID。此密钥由阿里云颁发，用于安全访问 Tablestore 资源。
     - **访问密钥**：与 AccessKey ID 相关联的用于认证的 AccessKey Secret。
     - **存储模型类型**：目前仅支持 `时序`。
   - 配置 TLS 参数。Tablestore 使用 HTTPS 端点，故默认情况下 TLS 已经启用，并且无需配置 TLS 参数。有关 TLS 连接选项的详细信息，请参阅[在专有版中配置 TLS/SSL](../deployments/tls_ssl.md)。
5. 在点击**创建**之前，您可以点击**测试连接**，以测试连接器是否能够连接到 Tablestore 服务器。
6. 点击最下方的**创建**按钮完成连接器的创建。在弹出的对话框中，您可以返回连接器列表，或继续创建规则和 Sink，以指定要转发到 Tablestore 的数据。具体步骤请参见[创建 Tablestore Sink 规则](#创建-tablestore-sink-规则)。

## 创建 Tablestore Sink 规则

本节演示了如何在 EMQX Cloud 中创建一条规则，用于处理来自源 MQTT 主题 `t/#` 的消息，并通过配置的 Sink 将处理后的结果发送到 Tablestore。

1. 登录 EMQX Cloud，进入部署，点击**数据集成** > **规则**。
2. 在规则页面点击右上角的**创建**按钮。
3. 输入规则 ID `my_rule`。
4. 在 SQL 编辑器中输入规则，例如将 `t/#` 主题的 MQTT 消息存储至 Tablestore，可以输入以下 SQL 语句：

   ```sql
   SELECT
     *
   FROM
     "t/#"
   ```

   > 如果您希望指定自己的 SQL 规则，必须确保规则选择出来的字段（`SELECT` 部分）包含之后在 Sink 中指定的 Tablestore 数据写入格式中包含的所有变量。

   如果您初次使用 SQL，可以点击 **SQL 示例**和**启用调试**来学习和测试规则 SQL 的结果。

5. 点击右侧的**添加动作**按钮，为规则在被触发的情况下指定一个动作。通过这个动作，EMQX Cloud 会将经规则处理的数据转发到 Tablestore。
6. 在动作类型下拉框中选择 **Alibaba 表格存储**，将动作下拉框保留为默认的**创建动作**。您也可以选择一个之前已经创建好的 Tablestore Sink。本示例将创建一个新的 Sink。
7. 为 Sink 输入一个名称。名称应结合使用大写/小写字母和数字。
8. 从连接器下拉框中选择之前创建的 `my_tablestore`。您也可以通过点击下拉框旁边的按钮创建一个新的连接器。有关配置参数，请参见[创建连接器](#创建连接器)。
9. 配置以下字段：

   - **数据源**：EMQX Cloud 从中获取消息的数据源，表示正在处理的数据的来源。可以是特定的主题或数据流。本示例使用 `mqtt`。
   - **表名**：数据将存储到 Tablestore 的表名。输入之前创建的表名 `telemetry`。您还可以使用变量（如 `${table}`）动态分配表名。
   - **度量名称**：在 Tablestore 中使用的度量名称，通常对应于数据的逻辑分组或类别。例如，它可以是类似 `temperature_readings` 或 `sensor_data` 的名称。您还可以使用变量（如 `${measurement}`）动态分配度量名称。本示例使用 `sensor_data`。
   - **存储模型类型**：Tablestore 使用的数据存储模型类型。目前仅支持 `timeseries`，优化用于基于时间的数据。
   - **标签**：标签是与每个数据项关联的键值对，可以用来添加元数据或标签，以便更容易查询和过滤。您可以点击**添加**来定义多个标签，例如：

     | 键 | 值 |
     | --- | --- |
     | `location` | `office1` |
     | `device` | `sensor1` |

   - **字段**：字段列表指定哪些数据将发送到 Tablestore。每个字段都映射到 Tablestore 表中的一列。您可以点击**添加**来添加以下内容：
     - **列**：Tablestore 中的列名。列名可以用变量，如 `${column_name}`。本示例使用 `temperature`。
     - **消息的值**：分配给列的值。该值可以是动态引用（如 `${payload.temperature}`）、布尔值（`true`）、数字（`1.3`）或二进制数据。本示例使用 `${payload.temperature}`。
     - **是否为整数**：如果列是数值类型，EMQX Cloud 默认按照浮点类型插入 Tablestore。如果想要插入整数值，需要将此标志设置为 `true`。
     - **是否为二进制**：若该列为二进制，EMQX Cloud 默认按照字符串类型插入 Tablestore。如果想要插入二进制数据，需要将此标志设置为 `true`。
   - **时间戳**：Tablestore 中记录的时间戳，整数值，单位为微秒。表示要插入到 Tablestore 的时间戳。您可以指定一个固定值，或者填写 `NOW` 字符串来表示希望 EMQX Cloud 动态填入处理消息时的当前时间，也可以使用变量占位符（如 `${microsecond_timestamp}`）动态分配。
   - **元数据更新模式**：定义 Tablestore 中元数据的更新策略：
     - `MUM_IGNORE`：忽略元数据更新，即使存在冲突更新，也确保元数据保持不变。
     - `MUM_NORMAL`：正常更新元数据，若元数据不存在，则在写入数据前动态创建元数据；若与现有元数据发生冲突，则可能会覆盖。
10. **备选动作（可选）**：如果您希望在消息投递失败时提升系统的可靠性，可以为 Sink 配置一个或多个备选动作。当 Sink 无法成功处理消息时，这些备选动作将被触发。更多信息请参见[备选动作](./actions.md#备选动作)。
11. 展开**高级设置**，根据需要配置高级设置选项（可选），详细请参考[高级设置](#高级设置)。
12. 在点击**创建**之前，您可以点击**测试连接**，以测试 Sink 是否能够连接到 Tablestore 服务器。
13. 点击**创建**完成 Sink 的创建。回到创建规则页面，您将看到新的 Sink 出现在动作输出页签下。
14. 在创建规则页面，验证配置的信息。点击**创建**按钮生成规则。

现在您已成功创建规则，您可以在规则页面上看到新的规则。点击**动作（Sink）**标签，您可以看到新的 Tablestore Sink。

## 测试规则

1. 使用 MQTTX 向 `t/1` 主题发布消息：

   ```bash
   mqttx pub -i emqx_c -t t/1 -m '{"temperature": 29.4}'
   ```

2. 查看规则和 Sink 运行统计，确认命中、发送成功次数均增加。
3. 前往 [Tablestore 控制台](https://otsnext.console.aliyun.com/)查看数据是否已经写入 Tablestore 中。输入度量名称 `sensor_data`，在标签中使用 `location=office1`、`device=sensor1` 作为查询条件，点击**查询**。

![在 Tablestore 中查询 EMQX Cloud 写入的数据](./_assets/tablestore-query-result.png)

## 高级设置

本节将深入介绍可用于 Tablestore 连接器和 Sink 的高级配置选项。在 EMQX Cloud 中配置连接器和 Sink 时，您可以根据您的特定需求展开**高级设置**，调整以下参数。

| 字段名称 | 描述 | 默认值 |
| --- | --- | --- |
| 缓存池大小 | 指定缓冲区工作进程数量。这些工作进程将被分配用于管理 EMQX Cloud 与 Tablestore 的出口 （egress）类型 Sink 中的数据流，它们负责在将数据发送到目标服务之前临时存储和处理数据。此设置对于优化性能并确保出口（egress）场景中的数据传输顺利进行尤为重要。对于仅处理入口 （ingress）数据流的桥接，此选项可设置为 `0`，因为不适用。 | `16` |
| 请求超期 | “请求 TTL”（生存时间）配置设置指定了请求在进入缓冲区后被视为有效的最长持续时间（以秒为单位）。此计时器从请求进入缓冲区时开始计时。如果请求在缓冲区内停留的时间超过此 TTL 设置，或者如果请求已发送但未能在 Tablestore 中及时收到响应或确认，则将视为请求已过期。 | `45` |
| 健康检查间隔 | 指定 Sink 将对与 Tablestore 的连接执行自动健康检查的时间间隔（以秒为单位）。 | `15` |
| 缓存队列最大长度 | 指定可以由 Tablestore Sink 中的每个缓冲区工作进程缓冲的最大字节数。缓冲区工作进程在将数据发送到 Tablestore 之前会临时存储数据，充当处理数据流的中介以更高效地处理数据流。根据系统性能和数据传输要求调整该值。 | `1` |
| 批量请求大小 | 指定可以在单个传输操作中从 EMQX Cloud 发送到 Tablestore 的数据批处理的最大大小。通过调整此大小，您可以微调 EMQX Cloud 与 Tablestore 之间数据传输的效率和性能。 | `1` |
| 请求模式 | 允许您选择`同步`或`异步`请求模式，以根据不同要求优化消息传输。在异步模式下，写入到 Tablestore 不会阻塞 MQTT 消息发布过程。但是，这可能导致客户端在它们到达 Tablestore 之前就收到了消息。 | `异步` |
| 请求飞行队列窗口 | “飞行队列请求”是指已启动但尚未收到响应或确认的查询。此设置控制 Sink 与 Tablestore 通信时可以同时存在的最大飞行队列请求数。当请求模式设置为`异步`时，“请求飞行队列窗口”参数变得特别重要。如果对于来自同一 MQTT 客户端的消息严格按顺序处理很重要，则应将此值设置为 `1`。 | `100` |