各行业连续查询实战

2026年7月10日 · 阅读需 16 分钟

LMGateway Staff Engineer

大多数人接触连续查询（Continuous Queries）是从降采样开始的：把每秒的数据聚合成每小时的平均值，写入小表，然后查询这个小表。但这只发挥了连续查询 10% 的能力。

在 IotEdgeDB 中，连续查询是一个按计划运行的 SQL 查询：它从源测量（source measurement）读取数据，将预计算结果写入目标测量（destination measurement）。查询使用标准的 DuckDB SQL——也就是 IotEdgeDB 即席分析所用的同一套引擎。这意味着，连续查询可以预计算 DuckDB 能表达的任何内容。

所有这些都在调度器上按计划执行，物化到轻量物化表中，仪表盘查询时只需毫秒级响应。

本文将深入五个行业场景：物联网、车联网、金融、制造与航空航天。每个场景都包含：模式（schema）、数据接入方式、典型查询和"进阶"查询。

30 秒速览

你只需一次 API 调用就能创建一个连续查询：

curl -X POST http://localhost:8080/api/v1/continuous_queries/ \
  -H "Authorization: Bearer <admin-token>" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "ohlc_1m",
    "database": "market_data",
    "source_measurement": "trades",
    "destination_measurement": "trades_ohlc_1m",
    "interval": "1m",
    "query": "SELECT time_bucket('1m', time) AS bucket, symbol, ... FROM market_data.trades WHERE time BETWEEN '{start_time}' AND '{end_time}' GROUP BY bucket, symbol",
    "retention_days": 90,
    "is_active": true
  }'

请求体字段说明：

字段	类型	必填	说明
`name`	string	是	唯一名称
`database`	string	是	源/目标数据库名
`source_measurement`	string	是	源测量名
`destination_measurement`	string	是	目标测量名（需以字母开头，仅含字母/数字/下划线/连字符）
`query`	string	是	聚合 SQL，必须包含 `{start_time}` 和 `{end_time}` 占位符
`interval`	string	是	调度间隔，如 `1m`、`5m`、`1h`
`retention_days`	int*	否	目标测量保留天数
`delete_source_after_days`	int*	否	处理多少天后删除源数据
`is_active`	bool	否	是否启用（默认 true）
`description`	string*	否	备注

两个占位符在执行时会被填充：{start_time} 和 {end_time} 定义当前处理的时间窗口。源表通过 database.measurement 全名引用（如 FROM market_data.trades），IotEdgeDB 在执行时会将该引用自动解析为对应存储后端的 read_parquet(...) 路径（本地磁盘、S3 或对象存储）。IotEdgeDB 会记录 last_processed_time，确保每次执行都能精确地从上次中断处继续，不重不漏。

所有连续查询端点都需要 Admin 权限（RequireAdmin 中间件）。完整端点清单：

方法	路径	说明
`POST`	`/api/v1/continuous_queries/`	创建
`GET`	`/api/v1/continuous_queries/?database=&is_active=true`	列表（支持按数据库、启停状态过滤）
`GET`	`/api/v1/continuous_queries/:id`	获取单个
`PUT`	`/api/v1/continuous_queries/:id`	更新
`DELETE`	`/api/v1/continuous_queries/:id`	删除
`POST`	`/api/v1/continuous_queries/:id/execute`	手动执行（body 可传 `start_time`/`end_time`/`dry_run`）
`GET`	`/api/v1/continuous_queries/:id/executions?limit=50`	执行历史

在 IotEdgeDB 社区版中，你需要手动触发执行（非常适合回填数据）；自动调度是企业版功能。最小调度间隔为 10 秒，每次执行超时时间为 10 分钟。

连续查询

在 IotEdgeDB Explorer 的连续查询模块中，你可以通过图形化界面创建和管理这些任务：填写 SQL、设置调度间隔、查看执行历史。无需写 cron，无需维护外部脚本。

物联网：预测性维护区间与漂移检测

工业物联网是降采样的起点，也是最显而易见的应用场景——但正因如此，值得展示一下超越显而易见的部分。目标不是一张更小的表，而是在轴承失效之前抓住它。

模式与数据接入

每台设备的振动和温度数据，按设备和区域打标签：

sensors,device=pump_07,zone=north vibration_mm_s=2.1,temp_c=58.4 1719300000000000000

Telegraf 配合 IotEdgeDB 原生输出可以直接写入 MessagePack 端点。

典型查询：异常区间与"最差设备"

一个 5 分钟的汇总查询，携带自身的异常区间（当前值相对于历史均值和标准差），并通过 ARG_MAX 标出每个区域中最差的单台设备：

SELECT
  time_bucket('5m', time) AS bucket,
  zone,
  AVG(vibration_mm_s) AS avg_vib,
  STDDEV(vibration_mm_s) AS sigma_vib,
  AVG(vibration_mm_s) + 3 * STDDEV(vibration_mm_s) AS upper_band,
  ARG_MAX(device, vibration_mm_s) AS worst_device
FROM iot.sensors
WHERE time BETWEEN '{start_time}' AND '{end_time}'
GROUP BY bucket, zone

worst_device 这一列才是值班人员真正想要的——不是"北区振动偏高"，而是"pump_07 是问题设备"。

进阶查询：基于回归的漂移检测

轴承失效通常不是突然尖峰，而是逐渐漂移。信号是振动随时间变化的斜率呈上升趋势。REGR_SLOPE 对（时间，数值）拟合一条直线，直接给出斜率。作为小时级连续查询运行：

SELECT
  time_bucket('1h', time) AS bucket,
  device,
  REGR_SLOPE(EXTRACT(EPOCH FROM time), vibration_mm_s) AS vib_trend_per_sec,
  CASE
    WHEN REGR_SLOPE(EXTRACT(EPOCH FROM time), vibration_mm_s) > 0.001 THEN 'degrading'
    ELSE 'normal'
  END AS health
FROM iot.sensors
WHERE time BETWEEN '{start_time}' AND '{end_time}'
GROUP BY bucket, device

物化后的 health 列把维护仪表盘变成了早期预警系统。一个正值且不断增长的 vib_trend_per_sec 是失效的早期形态——设备正在漂出其正常区间，远早于触发硬阈值之前。连续查询标记了这个问题，而没有人写过任何告警代码。

（当设备静默导致区间缺失时，查询中使用 generate_series 左连接来填补空白，确保趋势线不会因数据不连续而失真。）

车联网：车队遥测与能耗异常检测

在车联网场景中，数千辆网联车辆以秒级频率上报遥测数据——GPS 坐标、车速、电池 SOC、电机温度、胎压。对于车队运营商来说，真正的挑战不在于"收到数据"，而在于从海量数据流中实时提取可行动的洞察：哪辆车电池衰减加速？哪条路线能耗异常？某个区域是否出现批量车速骤降？

连续查询是车联网数据管道的核心引擎——它把原始遥测流转化为车队健康指标，预计算完成，仪表盘零延迟。

模式与数据接入

每辆车的遥测数据通过车载 T-Box 经 MQTT 上报，IotEdgeDB 的 MQTT 订阅功能直接写入时序库。按车辆和车队打标签，速度、SOC、GPS 坐标作为字段：

vehicle_telemetry,vehicle=EV-0247,fleet=delivery speed=58.3,soc=72.4,lat=39.9042,lon=116.4074 1719300000000000000

一笔数据包含了车辆状态的核心快照。配合 IotEdgeDB 的 MQTT 订阅 + 连续查询组合拳，数据从边缘设备直达物化分析表，端到端延迟可控制在秒级。

典型查询：车队实时聚合

按 5 分钟分桶，按车队维度汇总关键指标——活跃车辆数、平均速度、平均 SOC 以及车队内 SOC 的最大差距（用于识别离群车辆）：

SELECT
  time_bucket('5m', time) AS bucket,
  fleet,
  COUNT(DISTINCT vehicle) AS active_vehicles,
  AVG(speed) AS avg_speed,
  AVG(soc) AS avg_soc,
  MAX(soc) - MIN(soc) AS soc_range,
  ARG_MIN(vehicle, soc) AS lowest_soc_vehicle
FROM fleet.vehicle_telemetry
WHERE time BETWEEN '{start_time}' AND '{end_time}'
GROUP BY bucket, fleet

lowest_soc_vehicle 让调度员一眼看到"哪辆车需要优先充电"，无需打开子查询。

进阶查询：能耗效率与电池衰减检测

这是预测性维护在电动车队上的直接应用。电池健康不是看当前 SOC，而是看 SOC 随时间下降的速率——同一段路线、同样的载重，如果能耗速率持续上升，说明电池或者传动系统出现了渐变问题。

利用 REGR_SLOPE 在小时窗口内拟合（时间，SOC）的线性回归斜率，结合平均车速做分层判断：

SELECT
  time_bucket('1h', time) AS bucket,
  vehicle,
  AVG(speed) AS avg_speed,
  REGR_SLOPE(EXTRACT(EPOCH FROM time), soc) AS soc_drop_per_sec,
  CASE
    WHEN REGR_SLOPE(EXTRACT(EPOCH FROM time), soc) < -0.002 THEN 'battery_degrading'
    WHEN AVG(speed) > 80 AND REGR_SLOPE(EXTRACT(EPOCH FROM time), soc) < -0.001 THEN 'high_speed_abnormal_drain'
    ELSE 'normal'
  END AS energy_health
FROM fleet.vehicle_telemetry
WHERE time BETWEEN '{start_time}' AND '{end_time}'
GROUP BY bucket, vehicle

物化后的 energy_health 列让车队经理在问题变成抛锚事故之前就收到预警。更进一步，将 soc_drop_per_sec 与路线的海拔变化数据做 CORR 关联，可以区分"电池衰减"和"上坡自然消耗"，避免误报。

金融：OHLC K 线与 VWAP

原始市场数据是海量的逐笔成交记录。没有人会直接绘制逐笔成交图表——大家画的是 K 线：每个时间区间的开盘价、最高价、最低价、收盘价，再加上成交量和成交量加权平均价（VWAP）。

每次加载图表时，在数百万笔逐笔数据上实时计算这些指标，纯属浪费 CPU。正确的做法是：预计算。

模式与数据接入

一笔成交是一个测量（measurement），用 symbol 作为标签（tag），price 和 size 作为字段（field）：

trades,symbol=AAPL price=152.30,size=100 1719300000000000000

目标数据库通过请求头传递，因此 Line Protocol 主体保持简洁。生产环境建议使用 MessagePack 端点以提高吞吐量，但 Line Protocol 能更清晰地展示数据结构。

典型查询：1 分钟 K 线

这是最基础的操作。开盘价和收盘价分别是该时间桶内按时间排序的第一笔和最后一笔成交——这正是 ARG_MIN 和 ARG_MAX 的用途（取某列在另一列为最小值/最大值时所在行的值）：

SELECT
  time_bucket('1m', time) AS bucket,
  symbol,
  ARG_MIN(price, time) AS open,
  MAX(price) AS high,
  MIN(price) AS low,
  ARG_MAX(price, time) AS close,
  SUM(size) AS volume,
  SUM(price * size) / SUM(size) AS vwap
FROM market_data.trades
WHERE time BETWEEN '{start_time}' AND '{end_time}'
GROUP BY bucket, symbol

把这个查询放入 query 字段，设置 "interval": "1m"，目标表设为 trades_ohlc_1m。你的图表层现在读取的已经是 K 线数据了。

ARG_MIN/ARG_MAX 的精妙之处在于：它们在一次扫描中按时间戳选出开盘和收盘价，无需窗口函数，也无需自连接。

进阶查询：已实现波动率与分层 K 线

这里有两个进阶操作。

第一，已实现波动率。 一旦有了每分钟的收盘价，就可以用对数收益率在一个滑动窗口上的标准差来估计波动率。先用 CTE 计算对数收益率，再计算其滚动标准差。DuckDB 不允许窗口函数嵌套，所以两个步骤需要分开写。这可以作为第二层连续查询运行，其源表是 trades_ohlc_1m：

WITH log_returns AS (
  SELECT
    bucket,
    symbol,
    LN(close / LAG(close, 1) OVER (PARTITION BY symbol ORDER BY bucket)) AS ret
  FROM trades_ohlc_1m
)
SELECT
  bucket,
  symbol,
  STDDEV(ret) OVER (PARTITION BY symbol ORDER BY bucket ROWS BETWEEN 19 PRECEDING AND CURRENT ROW) AS realized_vol
FROM log_returns

注意：第二层 CQ 的源表是 trades_ohlc_1m（同样是 database.measurement 形式，如 FROM market_data.trades_ohlc_1m），查询中同样需要包含 {start_time}/{end_time} 占位符。

第二，分层 K 线。 连续查询的源可以是另一个连续查询的目标，因此你可以构建一个级联：原始成交 trades → trades_ohlc_1m → trades_ohlc_1h → trades_ohlc_1d。每一层重新聚合下一层的数据（开盘价取周期内第一笔，收盘价取最后一笔，最高/最低取最大/最小，成交量求和）。分钟 K 线用于日内交易，日 K 线用于多年走势图——全部自动维护，各自拥有独立的调度间隔和保留策略。

制造：SPC 控制限与 Cpk

一条生产线每个班次会产出数千次测量数据：孔径、灌装重量、扭矩等。质量工程师不关心原始数据流，他们关心的是统计过程控制（SPC）——均值、离散程度，以及判断过程是否受控的控制限（均值 ± 3σ）。这正是连续查询的典型应用场景。

模式与数据接入

按产线和工位打标签，测量值作为一个字段：

measurements,line=A,station=03 value=12.07 1719300000000000000

繁忙的生产线建议使用 MessagePack 端点。以下是通过 IotEdgeDB SDK 展示的列式结构：

client.write_points([
    {
        "measurement": "measurements",
        "tags": {"line": "A", "station": "03"},
        "fields": {"value": 12.07},
        "time": 1719300000000000000
    }
])

典型查询：各工位控制限

按 5 分钟分桶，计算均值和标准差，在同一 SELECT 中推导出上下控制限：

SELECT
  time_bucket('5m', time) AS bucket,
  line,
  station,
  AVG(value) AS mean,
  STDDEV(value) AS sigma,
  AVG(value) - 3 * STDDEV(value) AS lcl,
  AVG(value) + 3 * STDDEV(value) AS ucl
FROM factory.measurements
WHERE time BETWEEN '{start_time}' AND '{end_time}'
GROUP BY bucket, line, station

质量团队盯着看的控制图现在就是一条简单的 SELECT * FROM spc_5m，控制限已经算好了。

进阶查询：过程能力与根因相关性

质量部门真正上报的指标是 Cpk——相对于工程公差的过程能力指数。假设规格为 12.00 ± 0.15，Cpk 就是均值到最近规格限的距离除以 3σ。这只是对相同聚合数据的纯算术运算，可以直接纳入汇总查询：

SELECT
  time_bucket('1h', time) AS bucket,
  line,
  station,
  AVG(value) AS mean,
  STDDEV(value) AS sigma,
  (12.15 - AVG(value)) / (3 * STDDEV(value)) AS cpk_upper,
  (AVG(value) - 11.85) / (3 * STDDEV(value)) AS cpk_lower,
  CORR(temp_c, value) AS temp_corr
FROM factory.measurements
WHERE time BETWEEN '{start_time}' AND '{end_time}'
GROUP BY bucket, line, station

最后一列 CORR(temp_c, value) 才是真正的价值所在——它预计算了环境温度与每个工位测量值之间的小时级相关性。当某个工位开始出现漂移时，你不需要到处排查，物化表已经告诉你温度是否与之相关。

航空航天：变化率与 ASOF 流对齐

飞行器和平台遥测数据具有高速率、多通道的特点，而且很少在同一时钟下采样。发动机总线、机体 IMU 和航空电子设备各有各的采样频率。真正的难点不在于数据量，而在于计算那些时钟不同步的跨流衍生指标。连续查询可以提前完成这项工作，一次计算，多次使用。

模式与数据接入

两个测量，两个采样频率，同一架飞机：

airframe,aircraft=N12345 altitude=35000,airspeed=450 1719300000000000000
engine,aircraft=N12345 egt=720,fuel_flow=2400 1719300001000000000

典型查询：爬升率

爬升率是高度对时间的导数——相邻样本的高度变化除以时间间隔。LAG 取上一行，算术计算斜率：

SELECT
  time,
  aircraft,
  altitude,
  (altitude - LAG(altitude, 1) OVER (PARTITION BY aircraft ORDER BY time)) /
  (EXTRACT(EPOCH FROM time - LAG(time, 1) OVER (PARTITION BY aircraft ORDER BY time)) / 60) AS climb_rate_fpm
FROM aviation.airframe
WHERE time BETWEEN '{start_time}' AND '{end_time}'

这样就物化出了垂直速度（英尺/分钟），而航空电子设备从未发送过这个值。

进阶查询：跨采样频率的 ASOF 连接

这是在读取路径上很难低成本完成的操作。你想要的是：对于每一个机体采样点，获取该时刻最近的一次发动机读数——即使两个数据流走的是不同的时钟。

这就是 ASOF 连接。由于连续查询支持完整的 DuckDB 方言，你可以直接表达并物化对齐结果：

SELECT
  a.time,
  a.aircraft,
  a.altitude,
  a.airspeed,
  e.egt,
  e.fuel_flow
FROM aviation.airframe a
ASOF LEFT JOIN aviation.engine e
  ON a.aircraft = e.aircraft
  AND a.time >= e.time
WHERE e.time BETWEEN '{start_time}' - INTERVAL '1s' AND '{end_time}'
  AND a.time BETWEEN '{start_time}' AND '{end_time}'

连续查询只有一个 source_measurement，这里就是 airframe，通过占位符解析为其 Parquet 路径。第二个流（engine）通过完整的 database.measurement 名称显式引用。注意 e.time 的谓词条件：如果没有它，每次执行都会重新扫描整个发动机历史。将其限制在当前时间窗口（外加一个小的回看窗口，以便第一个机体采样点仍能找到它之前的发动机读数）可以保持每次执行的增量特性。

ASOF 连接是 DuckDB 的特性，而非 IotEdgeDB 独有的扩展。但因为 IotEdgeDB 通过 DuckDB 执行查询，DuckDB 支持的任何功能都可以在连续查询中使用——包括本文后面会提到的 ENTROPY 和 REGR_* 函数。

目标测量现在是一个单一的时间对齐表，每一行都包含机体和发动机的状态。跨流分析（"在这个高度和空速下 EGT 是多少？"）变成了对一张平表的扫描，而不是对两个数据洪流的连接操作。对齐成本只在调度上支付一次。

背后的统一模式

剥去行业外衣，每个示例都遵循同样的三步结构：

原始测量 → DuckDB 计算（按调度）→ 轻量物化表

计算步骤才是杠杆所在。它不限于平均值，因为它不是一个降采样功能——而是一个完整分析引擎的调度切片。OHLC、Cpk、ASOF 对齐、回归斜率、熵：只要你能用 DuckDB SQL 写出来，就能让它持续运行。

而且，因为连续查询的目标可以是另一个连续查询的源，你可以构建分层架构：

层级	粒度	调度间隔	典型保留期
原始数据	每条事件	—	1–7 天
第一层	1 分钟 / 5 分钟	1 分钟 / 5 分钟	30–90 天
第二层	1 小时	1 小时	90–365 天
第三层	1 天	1 天	1–5 年

每一层都有自己的调度间隔和保留策略。配合连续查询的 retention_days 参数和原始测量的保留策略，数据生命周期自动运行：高分辨率数据保留数天，能回答 99% 查询的汇总数据保留数年，存储成本与实际保留的数据量挂钩。

总结

连续查询的价值不在于"自动执行 SQL"这件事本身——cron job 也能做到。它的价值在于物化计算结果的数据库内集成：

计算与存储一体：源读 Parquet，目标写 Parquet，零序列化开销
增量执行：{start_time} / {end_time} 占位符保证不重不漏，精确接续
DuckDB 全方言支持：窗口函数、ASOF 连接、统计回归、熵函数，不受限制
级联分层：分钟→小时→天，逐层聚合，长期保留
一次计算，处处使用：仪表盘、告警、导出——全部读同一张物化表

30 秒速览​

物联网：预测性维护区间与漂移检测​

模式与数据接入​

典型查询：异常区间与"最差设备"​

进阶查询：基于回归的漂移检测​

车联网：车队遥测与能耗异常检测​

模式与数据接入​

典型查询：车队实时聚合​

进阶查询：能耗效率与电池衰减检测​

金融：OHLC K 线与 VWAP​

模式与数据接入​

典型查询：1 分钟 K 线​

进阶查询：已实现波动率与分层 K 线​

制造：SPC 控制限与 Cpk​

模式与数据接入​

典型查询：各工位控制限​

进阶查询：过程能力与根因相关性​

航空航天：变化率与 ASOF 流对齐​

模式与数据接入​

典型查询：爬升率​

进阶查询：跨采样频率的 ASOF 连接​

背后的统一模式​

总结​

30 秒速览

物联网：预测性维护区间与漂移检测

模式与数据接入

典型查询：异常区间与"最差设备"

进阶查询：基于回归的漂移检测

车联网：车队遥测与能耗异常检测

模式与数据接入

典型查询：车队实时聚合

进阶查询：能耗效率与电池衰减检测

金融：OHLC K 线与 VWAP

模式与数据接入

典型查询：1 分钟 K 线

进阶查询：已实现波动率与分层 K 线

制造：SPC 控制限与 Cpk

模式与数据接入

典型查询：各工位控制限

进阶查询：过程能力与根因相关性

航空航天：变化率与 ASOF 流对齐

模式与数据接入

典型查询：爬升率

进阶查询：跨采样频率的 ASOF 连接

背后的统一模式

总结