EMQX Neuron

简体中文

English

简体中文

English

Appearance

联系我们
联系我们

Snowflake ​

采集到的点位数据要做长期存储和分析时，通常落到数据仓库。EMQX Neuron 不直接对接 Snowflake——它把数据发到 EMQX，由 EMQX 的数据集成写入：

现场设备 → EMQX Neuron → EMQX Cloud / EMQX Enterprise → Snowflake

这样分工的好处是边缘侧只维护一条 MQTT 链路。下游换成别的仓库、或者同时写多个仓库，都在 EMQX 侧配置，EMQX Neuron 不用动。

本文介绍链路两端怎么对齐：EMQX Neuron 发出什么样的报文，EMQX 侧的规则怎么把它映射成表的列。Snowflake 本身的配置以 EMQX 官方文档为准，文中给出对应链接。

前置条件 ​

  • 一套可用的 EMQX Cloud 部署或自建 EMQX Enterprise，EMQX Neuron 已连上并在上报数据，见 EMQX Cloud
  • Snowflake 账号，且有创建数据库、表、Stage 与 Pipe 的权限

TIP

数据集成是 EMQX Enterprise 与 EMQX Cloud 的功能，开源版 EMQX 不包含。

EMQX Neuron 发出的报文 ​

默认的 values-format 报文结构固定，这是后续所有映射的依据：

json
{
  "node": "modbus-tcp",
  "group": "group-1",
  "timestamp": 1790219933821,
  "values": { "temperature": 23.5, "pressure": 1013 },
  "errors": {},
  "metas": {}
}

默认主题为 /neuron/{应用名}/{驱动名}/{组名}，可在每条订阅中自定义。存在多个工厂或产线时建议按层级命名，例如 factory-a/line-1/modbus-tcp，EMQX 侧用通配符一次订阅全部，见数据上云 · 上报主题。

点位值在 values 这一层嵌套里，点位名就是键名。这一点决定了下面规则 SQL 的写法。

两种写入模式 ​

Snowflake 有两种写入模式：聚合模式先把消息攒成 CSV 文件上传到 Stage，再由 Snowpipe 加载进表；流式模式通过 Snowpipe Streaming API 逐行实时写入。点位数据量大、允许分钟级延迟时用聚合模式更省成本；要求秒级可见时用流式模式。

Snowflake 侧准备 ​

创建数据库、Schema 和接收表。表的列要和后面规则 SQL 选出的字段一一对应：

sql
CREATE DATABASE IF NOT EXISTS neuron_data;
CREATE SCHEMA   IF NOT EXISTS neuron_data.public;

CREATE TABLE IF NOT EXISTS neuron_data.public.telemetry (
  node                STRING,
  group_name          STRING,
  publish_received_at TIMESTAMP_NTZ,
  temperature         DOUBLE,
  pressure            DOUBLE
);

再创建 Stage、Pipe，以及一个具备管道操作权限的用户和角色。流式模式还需要为该用户配置 RSA 密钥对认证：

shell
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out snowflake_rsa_key.private.pem -nocrypt
openssl rsa -in snowflake_rsa_key.private.pem -pubout -out snowflake_rsa_key.public.pem

完整的对象创建语句与授权见 EMQX 文档的 Snowflake 数据集成。

在 EMQX 中创建连接器 ​

聚合模式选 Snowflake（ODBC），流式模式选 Snowflake（Streaming API）。主要字段：

字段取值
服务器地址org-account.snowflakecomputing.com
账户org-account，即组织 ID 与账户名
用户名前面创建的管道用户
密码或私钥路径聚合模式二选一；流式模式必须用私钥
启用 TLS流式模式必须开启

创建规则 ​

规则 SQL 负责把 EMQX Neuron 的嵌套报文摊平成表的列。

WARNING

不要用 SELECT *。 Snowflake Sink 要求选出的字段名与数量和目标表的列完全一致，多一列少一列都会写入失败。

sql
SELECT
  payload.node                                          as node,
  payload.group                                         as group_name,
  unix_ts_to_rfc3339(publish_received_at, 'millisecond') as publish_received_at,
  payload.values.temperature                            as temperature,
  payload.values.pressure                               as pressure
FROM
  "/neuron/#"

payload.values.温度点位名 就是取 EMQX Neuron 报文里的点位值。需要入库哪些点位，就在这里列哪些，并保证表里有同名的列。点位增减时，表结构和这条 SQL 要一起改。

添加 Sink ​

聚合模式选 Snowflake 类型，填数据库名、Schema、Stage、Pipe、管道用户与私钥，另外两个参数决定上传节奏：

参数说明
最大记录数攒够这么多条就上传一次
时间间隔距上次上传超过这么多秒就上传一次

两者谁先满足就触发。点位采集周期 1 秒、单组 20 个点位时，默认的 1000 条约合 50 秒一个文件。

流式模式选 Snowflake-Streaming 类型，填数据库名、Schema 和流式 Pipe 名即可，没有攒批参数。

延伸阅读 ​