补充业务上下文
在 Databricks 中将设备事件与维护、生产或业务记录关联分析，排查运营问题。
概览
让设备数据在 Databricks 中发挥价值
将遥测数据与业务数据结合，分析运营问题，构建用于数据分析和 AI 的数据集。
存储可直接使用的数据
写入文件之前，先在 EMQX 中过滤消息、整理消息负载，让湖仓只接收分析所需的字段。
数据接入与分析解耦
以 S3 作为共享存储层：EMQX 负责上传设备数据，Databricks 按自身的调度读取和处理文件。
架构
从 MQTT 消息到 Databricks 查询
EMQX 处理消息并将文件写入 Amazon S3，Databricks 通过 Unity Catalog 外部位置读取这些文件，用于 SQL 查询和后续处理。
- 采集
MQTT 设备
遥测数据与事件
- 处理与路由 EMQX
规则 + S3 Sink
过滤 · 转换 · 上传
- 存储
Amazon S3
指定的存储桶和前缀
- 分析
Databricks
外部位置 · SQL · AI
从信号到数据集
| 设备 | 温度 °C | 采集时间 |
|---|---|---|
| sensor-01 | 23.8 | 09:42:18 |
| sensor-02 | 24.1 | 09:42:21 |
| sensor-03 | 22.6 | 09:42:24 |
核心能力
精准控制进入湖仓的数据
过滤 MQTT 消息、选择文件格式，并为 Databricks 合理组织 S3 中的文件。
过滤与转换
使用 SQL 规则选择主题和消息负载字段，在 S3 Sink 写入之前过滤掉无用消息并转换记录。
选择格式与批处理方式
既可以逐条写入消息，也可以将记录批量写入 CSV、JSON Lines 或 Parquet 文件。每个批次可设置记录数或时间上限，可用格式取决于您的 EMQX 版本。
在 S3 中有序组织文件
使用支持的模板变量设置 S3 对象键，按时间或上传顺序对文件分组，让 Databricks 始终从一致的路径读取数据。
工作原理
四步完成数据接入配置
开始前，请准备好支持 S3 集成的 EMQX 部署、一个 S3 存储桶，以及启用了 Unity Catalog 的 Databricks on AWS 工作区。
- 01
配置 S3 访问权限
选择存储桶和前缀，然后为该 S3 路径创建 Databricks 存储凭据和外部位置。
- 02
将 EMQX 连接到 S3
使用所在区域的端点和 AWS 凭证创建 Amazon S3 连接器，然后测试连接。
- 03
定义规则与 Sink
创建 SQL 规则并添加 S3 Sink，设置对象键、文件格式和上传方式。
- 04
发布并验证
发布一条测试 MQTT 消息，在 S3 中检查生成的文件，再从 Databricks 读取该文件，确认数据和格式正确。
应用场景
从遥测数据中获取运营洞察
在 Databricks 中分析设备历史数据、对比各站点表现、准备机器学习数据。
设备趋势分析
将温度、振动和状态历史与维护记录进行对比，及时发现设备运行状态的变化。
了解工业物联网方案多站点表现对比
将遥测数据与生产、能耗和位置数据结合，按站点、资产和时间段对比运行表现。
机器学习数据集
构建历史遥测数据集，用于在 Databricks 中进行特征工程、模型训练和评估。
FAQ
常见问题
本页介绍的方案以 Amazon S3 作为中转。EMQX 处理 MQTT 消息后，通过 S3 Sink 写入文件；Databricks 再通过 Unity Catalog 外部位置读取这些文件。
直接上传会按您配置的内容存储数据。聚合上传支持 CSV、JSON Lines 和 Parquet，具体取决于您的 EMQX 版本。请选择 Databricks 工作流能够读取的格式，并使用有代表性的消息进行测试。
数据时效取决于上传批处理设置、文件写入 S3 的耗时，以及 Databricks 读取或处理文件的时机。使用聚合上传时，可以调整记录数和时间上限，在文件大小与数据时效之间取得平衡。
EMQX 需要拥有向所选 S3 存储桶和前缀写入数据的权限。Databricks 则需要基于 IAM 的存储凭据、外部位置，以及读取这些文件所需的 Unity Catalog 权限。
可以。在 S3 Sink 写入供 Databricks 读取的文件之前，EMQX 的 SQL 规则可以匹配主题、选取消息负载字段，并对消息进行过滤或转换。