将 MQTT 数据写入到 Microsoft SQL Server
TIP
Microsoft SQL Server 数据集成是 EMQX 企业版功能。
SQL Server 是领先的关系型商业数据库解决方案之一，被广泛应用于各种规模和类型的企业和组织中。EMQX 支持与 SQL Server 集成，使您能够将 MQTT 消息和客户端事件保存到 SQL Server 以便于构建复杂的数据管道和分析流程实现数据管理和分析，或进行设备连接管理并与其他 ERP, CRM，BI 企业系统的集成。
本页详细介绍了 EMQX 与 Microsoft SQL Server 的数据集成并提供了实用的规则和 Sink 创建指导。
TIP
仅 EMQX 5.0.3 及以上版本支持 Microsoft SQL Server Sink 功能。
工作原理
Microsoft SQL Server 数据集成是 EMQX 的开箱即用功能，结合了 EMQX 的设备接入、消息传输能力与 Microsoft SQL Server 强大的数据存储能力。通过内置的规则引擎组件和 Sink ，您可以将 MQTT 消息和客户端事件存储到 Microsoft SQL Server 中，也可以通过事件触发对 Microsoft SQL Server 中数据的更新或删除操作，从而实现对诸如设备在线状态、上下线历史等的记录。该集成简化了从 EMQX 到 Microsoft SQL Server 的数据摄取过程，无需复杂的编码。
下图展示了 EMQX 和 SQL Server 之间的数据集成的典型架构:
将 MQTT 数据摄取到 Microsoft SQL Server 的工作流程如下：
- 消息发布和接收：工业物联网设备通过 MQTT 协议成功连接到 EMQX，并根据其运行状态、读数或触发的事件，从机器、传感器和生产线发布实时 MQTT 数据到 EMQX。当 EMQX 接收到这些消息时，它将在其规则引擎中启动匹配过程。
- 消息数据处理：当消息到达时，它会通过规则引擎进行处理，然后由 EMQX 中定义的规则处理。规则根据预定义的标准确定哪些消息需要路由到 Microsoft SQL Server。如果任何规则指定了载荷转换，那么这些转换将被应用，例如转换数据格式、过滤出特定信息，或用额外的上下文丰富载荷。
- 数据写入到 Microsoft SQL Server：规则触发将消息写入 Microsoft SQL Server 的操作。借助 SQL 模板，用户可以从规则处理结果中提取数据来构造 SQL 并发送到 Microsoft SQL Server 执行，从而将消息的特定字段写入或更新到数据库的相应表和列中。
- 数据存储和利用：数据现存储在 Microsoft SQL Server 中，企业可以利用其查询能力应用于各种用例。
特性与优势
与 Microsoft SQL Server 的数据集成提供了一系列特性和优势，确保了数据传输、存储和利用的高效性：
- 实时数据流：EMQX 专为处理实时数据流而构建，确保了从源系统到 Microsoft SQL Server 的数据传输的高效性和可靠性。它使组织能够实时捕获和分析数据，非常适合需要立即洞察和行动的用例。
- 高性能和可扩展性：EMQX 和 Microsoft SQL Server 都具有扩展性和可靠性的特点，适用于处理大规模的物联网数据，并在需求增长时进行不停机的水平和垂直扩展，确保物联网应用程序的连续性和可靠性。
- 数据转换的灵活性：EMQX 提供了强大的基于 SQL 的规则引擎，允许组织在将数据存储到 Microsoft SQL Server 之前进行预处理。它支持各种数据转换机制，如过滤、路由、聚合和丰富，使组织能够根据他们的需求塑造数据。
- 高级分析：Microsoft SQL Server 提供了强大的分析能力，例如通过 Analysis Services 构建多维数据模型，以支持复杂的数据分析和数据挖掘，通过 Reporting Services 创建和发布报告，向利益相关者展示物联网数据的洞察和分析结果。
准备工作
本节介绍了在 EMQX 中创建 Microsoft SQL Server 数据集成之前需要做的准备工作，包括如何设置 Microsoft SQL Server 服务器并创建数据库和数据表、安装并配置 ODBC 驱动程序。
前置准备
安装并连接到 Microsoft SQL Server
本节描述如何使用 Docker 镜像在 Linux/MacOS 安装启动 Microsoft SQL Server 2019 以及如何使用 sqlcmd 连接到 Microsoft SQL Server。关于其他 Microsoft SQL Server 的安装方式，请参阅微软提供的 Microsoft SQL Server 安装指南。
通过 Docker 安装并启动 Microsoft SQL Server。
Microsoft SQL Server 要求使用复杂密码，请参阅使用复杂密码。 使用环境变量
ACCEPT_EULA=Y启动 Docker 容器代表您同意 Microsoft 的 EULA 条款，另请参阅 End-User Licensing Agreement。bash# 启动一个 Microsoft SQL Server 容器并设置密码为 `mqtt_public1` $ docker run --name sqlserver -p 1433:1433 -e ACCEPT_EULA=Y -e MSSQL_SA_PASSWORD=mqtt_public1 -d mcr.microsoft.com/mssql/server:2022-CU15-ubuntu-22.04进入 Docker 容器。
bash$ docker exec -it sqlserver bash在容器中连接到 Microsoft SQL Server 服务器，需要输入预设的密码。输入密码时字符不会回显。请输入密码后直接键入
Enter。bash$ /opt/mssql-tools18/bin/sqlcmd -S localhost -U sa -P mqtt_public1 -N -C 1>
TIP
Microsoft 提供的 Microsoft SQL Server 容器内已安装 mssql-tools18，但可执行文件并不在 $PATH 中。因此，在继续操作之前，您需要为 sqlcmd 指定可执行文件的路径。对于本例中的 Docker 部署，文件路径应为 /opt。
关于更多 mssql-tools18 的使用，请阅读 sqlcmd 实用工具。
至此 Microsoft SQL Server 2022 实例已经完成部署并可以连接。
创建数据表
使用已创建的连接和下面的 SQL 语句在 Microsoft SQL Server 中创建数据表。
如需用于 MQTT 消息存储，创建数据表
dbo.t_mqtt_msg。该表存储每条消息的 MsgID、主题、QoS、Payload 以及发布时间。sqlCREATE TABLE dbo.t_mqtt_msg (id int PRIMARY KEY IDENTITY(1000000001,1) NOT NULL, msgid VARCHAR(64) NULL, topic VARCHAR(100) NULL, qos tinyint NOT NULL DEFAULT 0, payload VARCHAR(100) NULL, arrived DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP); GO如需用于设备上下线状态记录，创建数据表
dbo.t_mqtt_events。sqlCREATE TABLE dbo.t_mqtt_events (id int PRIMARY KEY IDENTITY(1000000001,1) NOT NULL, clientid VARCHAR(255) NULL, event_type VARCHAR(255) NULL, event_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP); GO
安装并配置 ODBC 驱动程序
为了能够访问 Microsoft SQL Server 数据库，您需要安装并配置 ODBC 驱动程序。您可以使用 Microsoft 发布的 msodbcsql18 或者 FreeTDS 作为 ODBC 驱动程序。
EMQX 使用 odbcinst.ini 配置中的 DSN Name 来确定驱动动态库的路径，有关的详细信息请参考连接属性。
注意
您可以根据自己的喜好命名 DSN Name，但建议只使用英文字母。此外 DSN Name 大小写敏感。
安装配置 msodbcsql18 作为 ODBC 驱动程序
如需安装配置 msodbcsql18 作为 ODBC 驱动程序，您需要参考微软的安装指导：
受限于 Microsoft EULA 条款，EMQX 提供的 Docker 镜像不带有 msodbcsql18 驱动程序，如需在 Docker 或 Kubernetes 中使用该驱动程序，您需要基于 EMQX Enterprise 提供的镜像构建带有 ODBC 驱动程序的新镜像以便在连接 Microsoft SQL Server 数据库时使用 msodbcsql18 驱动程序。使用构建的新镜像，代表您同意 Microsoft SQL Server EULA 条款。
按照以下说明构建新镜像：
使用以下 Dockerfile 来构建新镜像。
本示例中的基础镜像版本为
emqx/emqx-enterprise:5.8.1。您可以根据所需的 EMQX Enterprise 版本来构建镜像，或者使用最新版本的镜像emqx/emqx-enterprise:latest。dockerfileFROM emqx/emqx-enterprise:5.8.1 USER root RUN apt-get -qq update && apt-get install -yqq curl gpg && \ . /etc/os-release && \ curl -fsSL https://packages.microsoft.com/keys/microsoft.asc | gpg --dearmor -o /usr/share/keyrings/microsoft-prod.gpg && \ curl -fsSL "https://packages.microsoft.com/config/${ID}/${VERSION_ID}/prod.list" > /etc/apt/sources.list.d/mssql-release.list && \ apt-get -qq update && \ ACCEPT_EULA=Y apt-get install -yqq msodbcsql18 unixodbc-dev && \ sed -i 's/ODBC Driver 18 for SQL Server/ms-sql/g' /etc/odbcinst.ini && \ apt-get clean && \ rm -rf /var/lib/apt/lists/* USER emqx使用命令
docker build -t emqx/emqx-enterprise:5.8.1-msodbc构建镜像。构建完成后可以使用
docker image ls来获取本地的 image 列表，您也可以将镜像上传或保存备用。
注意
使用上文给出的示例安装 msodbcsql18 驱动后，请确认 odbcinst.ini 中的 DSN Name 为 ms-sql 。您也可以根据需要修改 DSN Name。
安装配置 FreeTDS 作为 ODBC 驱动程序
本节介绍了在几种主流发行版上安装配置 FreeTDS 作为 ODBC 驱动程序的方式。在此处给出的示例中，DSN Name 均为 ms-sql。
在 MacOS 上安装配置 FreeTDS 作为 ODBC 驱动程序:
$ brew install unixodbc freetds
$ vim /usr/local/etc/odbcinst.ini
[ms-sql]
Description = ODBC for FreeTDS
Driver = /usr/local/lib/libtdsodbc.so
Setup = /usr/local/lib/libtdsodbc.so
FileUsage = 1在 Centos 上安装配置 FreeTDS 作为 ODBC 驱动程序:
$ yum install unixODBC unixODBC-devel freetds freetds-devel perl-DBD-ODBC perl-local-lib
$ vim /etc/odbcinst.ini
# 加入以下内容
[ms-sql]
Description = ODBC for FreeTDS
Driver = /usr/lib64/libtdsodbc.so
Setup = /usr/lib64/libtdsS.so.2
Driver64 = /usr/lib64/libtdsodbc.so
Setup64 = /usr/lib64/libtdsS.so.2
FileUsage = 1在 Ubuntu 上安装配置 FreeTDS 作为 ODBC 驱动程序（以 Ubuntu20.04 为例，其他版本请参考 ODBC 官方文档）:
$ apt-get install unixodbc unixodbc-dev tdsodbc freetds-bin freetds-common freetds-dev libdbd-odbc-perl liblocal-lib-perl
$ vim /etc/odbcinst.ini
# 加入以下内容
[ms-sql]
Description = ODBC for FreeTDS
Driver = /usr/lib/x86_64-linux-gnu/odbc/libtdsodbc.so
Setup = /usr/lib/x86_64-linux-gnu/odbc/libtdsS.so
FileUsage = 1创建连接器
在创建 Microsoft SQL Server Sink 之前，您需要创建一个连接器，以便 EMQX 与 Microsoft SQL Server 服务建立连接。以下示例假定您在本地机器上同时运行 EMQX 和 Microsoft SQL Server。如果您在远程运行 Microsoft SQL Server 和 EMQX，请相应地调整设置。
转到 Dashboard 集成 -> 连接器 页面。点击页面右上角的创建。
在连接器类型中选择 Microsoft SQL Server，点击下一步。
在 配置 步骤，配置以下信息：
- 连接器名称：应为大写和小写字母及数字的组合，例如：
my_sqlserver。 - 服务器地址：
127.0.0.1:1433，或使用实际的 Microsoft SQL Server 地址和端口 - 数据库名字：
master - 用户名：
sa - 密码：
mqtt_public1 - SQL Server Driver 名称：
ms-sql，即您在odbcinst.ini中配置的 DSN Name
- 连接器名称：应为大写和小写字母及数字的组合，例如：
高级配置（可选）：详细请参考 Sink 的特性。
在点击创建之前，您可以点击测试连接来测试连接器是否能连接到 Microsoft SQL Server 服务器。
点击创建按钮完成连接器创建。
在弹出的创建成功对话框中您可以点击创建规则，继续创建规则以指定需要写入 RocketMQ 的数据和需要记录的客户端事件。您也可以按照创建消息存储 Sink 规则和创建事件记录 Sink 规则章节的步骤来创建规则。
创建消息存储 Sink 规则
本节演示了如何在 Dashboard 中创建一条规则，用于处理来自源 MQTT 主题 t/# 的消息，并通过配置的 Sink 将处理后的结果写入到 Microsoft SQL Server 的数据表 dbo.t_mqtt_msg 中。
转到 Dashboard 集成 -> 规则页面。
点击页面右上角的创建。
输入规则 ID
my_rule。如需实现对指定主题消息的转发，例如将t/#主题的 MQTT 消息存储至 Microsoft SQL Server，在 SQL 编辑器中输入以下 SQL 语句： 注意：如果您希望制定自己的 SQL 语句，需要确保规则选出的字段（SELECT 部分）包含之后在 Sink 中配置的 SQL 模板中所有用到的变量。sqlSELECT * FROM "t/#"TIP
如果您初次使用 SQL，可以点击 SQL 示例和启用调试来学习和测试规则 SQL 的结果。
点击右侧的添加动作按钮，为规则在被触发的情况下指定一个动作。通过这个动作，EMQX 会将经规则处理的数据发送到 Microsoft SQL Server。
在动作类型下拉框中选择
Microsoft SQL Server，保持动作下拉框为默认的创建动作选项，您也可以选择一个之前已经创建好的 Microsoft SQL Server Sink。此示例将创建一个全新的 Sink 并添加到规则中。输入 Sink 名称，名称应为大/小写字母和数字的组合。
从连接器下拉框中选择刚刚创建的
my_sqlserver。您也可以通过点击下拉框旁边的按钮创建一个新的连接器。有关配置参数，请参见创建连接器。配置 SQL 模板。如需实现对指定主题消息的转发，使用如下 SQL 语句完成数据插入。此处为预处理 SQL，字段不应当包含引号，SQL 末尾不要带分号
;。sqlinsert into dbo.t_mqtt_msg(msgid, topic, qos, payload) values ( ${id}, ${topic}, ${qos}, ${payload} )如果在模板中使用未定义的占位符变量，您可以切换未定义变量作为 NULL 开关（位于 SQL 模板 上方）来定义规则引擎的行为：
关闭（默认）：规则引擎可以将字符串
undefined插入数据库。启用：允许规则引擎在变量未定义时将
NULL插入数据库。TIP
如果您初次使用 SQL，可以点击 SQL 示例 和启用调试来学习和测试规则 SQL 的结果。
高级配置（可选），根据情况配置同步/异步模式，队列与批量等参数，详细请参考 Sink 的特性。
在点击创建按钮完成 Sink 创建之前，您可以使用测试连接来测试当前 Sink 到 Microsoft SQL Server 的连接是否成功。
点击创建按钮完成 Sink 创建，新建的 Sink 将被添加到动作输出列表中。
回到创建规则页面，对配置的信息进行确认，点击创建。一条规则应该出现在规则列表中。
现在您已成功创建了通过 Microsoft SQL Server Sink 将数据转发到 Microsoft SQL Server 的规则，同时在规则页面的动作(Sink) 标签页看到新建的 Microsoft SQL Server Sink。
您还可以点击 集成 -> Flow 设计器可以查看拓扑，通过拓扑可以直观的看到，主题 t/# 下的消息在经过规则 my_rule 解析后被发送到 Microsoft SQL Server 中。
创建事件记录 Sink 规则
本节展示如何创建用于记录客户端上/下线状态的规则，并通过配置的 Sink 将记录写入到 Microsoft SQL Server 的数据表 dbo.t_mqtt_events 中。
注意：除规则 SQL 和 Sink 的 SQL 模板设置不同外，其他操作步骤与创建消息存储 Sink 规则章节完全相同。
规则 SQL 如下：
SELECT
*,
floor(timestamp / 1000) as s_shift,
timestamp div 1000 as ms_shift
FROM
"$events/client_connected", "$events/client_disconnected"Sink 的 SQL 模板如下:
insert into dbo.t_mqtt_events(clientid, event_type, event_time) values ( ${clientid}, ${event}, DATEADD(MS, ${ms_shift}, DATEADD(S, ${s_shift}, '19700101 00:00:00:000') ) )测试规则
使用 MQTTX 向 t/1 主题发布消息。
mqttx pub -i emqx_c -t t/1 -m '{ "msg": "Hello SQL Server" }'查看 Microsoft SQL Server Sink 运行统计。
- 用于消息存储的 Sink ，命中、发送成功次数均 +1。查看数据是否已经写入
dbo.t_mqtt_msg表中：
1> SELECT * from dbo.t_mqtt_msg
2> GO
id msgid topic qos payload arrived
----------- ---------------------------------------------------------------- ---------------------------------------------------------------------------------------------------- --- ---------------------------------------------------------------------------------------------------- -----------------------
1000000001 0005F995096D9466F442000010520002 t/1 0 { "msg": "Hello SQL Server" } 2023-04-18 04:49:47.170
(1 rows affected)
1>- 用于存储上下线事件的 Microsoft SQL Server Sink ，命中、发送次数均 +2，即一次上线和一次下线。查看设备状态是否已经写入
dbo.t_mqtt_events表中：
1> SELECT * from dbo.t_mqtt_events
2> GO
id clientid event_type event_time
----------- ---------------------------------------------------------------- ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- -----------------------
1000000001 emqx_c client.connected 2023-04-18 04:49:47.140
1000000002 emqx_c client.disconnected 2023-04-18 04:49:47.180
(2 rows affected)
1>