img

摘要

一、IoT 时序数据的五类"脏"问题

先把要处理的问题摆清楚。不同类型的脏数据，检测方法和清洗策略完全不同，混在一起处理只会越洗越乱。

类型	典型表现	根因	危害
缺值	时间轴上某段没有数据	传感器掉线、上报丢失	统计指标失真、模型输入断档
异常值	单点数值远超合理范围	传感器故障尖刺、电磁干扰	拉偏均值、触发误告警
重复值	同一时刻多条相同记录	网络重传、边缘端补传	count 虚高、聚合重复计算
卡死值	连续多个采样值完全相同	传感器卡死、通信队列冻结	制造"平稳"假象，掩盖真实波动
时序乱序	时间戳不单调递增	网络延迟、多源时钟不同步	窗口聚合错位、趋势计算反向

img

其中卡死值是工业场景特别需要单独拎出来的——它不像异常值那样"窜尖"，反而表现为"过于平稳"，很容易被当成正常数据忽略，但一段本该波动的振动信号突然变成直线，往往是传感器或采集链路出问题的信号。

下面逐类讲检测与清洗。

二、缺值：检测与三种填充策略

2.1 先检测，再决定填不填

缺值处理的第一步不是"怎么填"，而是"哪里缺、缺多久"。盲目填充会引入虚假数据，比缺值本身更有害。

检测采样断档，看相邻样本的时间间隔是否异常变大：

代码语言：SQL

复制

// 假设温度传感器正常采样间隔 1 秒，找出间隔超过 3 秒的断档gaps = selectdeviceId,ts as gap_start,prev(ts) as gap_before,(ts - prev(ts)) 1000as gap_seconds // 毫秒转秒from loadTable("dfs://iot", "temperature")context by deviceIdha ving (ts - prev(ts)) 1000 > 3

prev(ts) 配合 context by deviceId，取同一设备上一条记录的时间戳。相邻间隔超过阈值（这里设 3 秒，正常 1 秒的 3 倍）的位置就是断档点。

2.2 三种填充策略，各管一种场景

确认了缺值位置后，是否填充、用什么方式填充，要看断档长度和业务对连续性的要求：

代码语言：SQL

复制

// 策略一：前向填充（ffill）—— 用上一个有效值顶替// 适合：短断档、慢变量（温度、压力），值变化平缓select deviceId, ts, ffill(temperature) as temperature_filledfrom loadTable("dfs://iot", "temperature")context by deviceId// 策略二：不填充，显式置空// 适合：长断档（超过物理合理时间）、快变量（振动），填充会制造假数据// 后续聚合用 a vg 等会自动忽略空值，不污染统计// 策略三：限定填充长度，短断档填、长断档不填// ffill 填充后，再用断档标记把长断档段重新置空

img

核心判断：缺值要不要填，取决于"这段缺失的数据，用邻近值顶替会不会误导下游"。慢变量（温度每秒变化零点几度）短断档用前向填充问题不大；快变量（振动）一旦断档，邻近值根本代表不了缺失段，强行填充等于编造数据。

一个实用经验：给填充后的数据打一个"是否为填充值"的标记列，让下游知道哪些是真实采样、哪些是补的，避免被填充值误导。

代码语言：SQL

复制

// 带填充标记的清洗结果cleaned = selectdeviceId, ts,temperature as temperature_raw,ffill(temperature)as temperature_filled,isNull(temperature) as is_imputed// 1=填充值, 0=真实采样from loadTable("dfs://iot", "temperature")context by deviceId

三、异常值：从 3-sigma 到鲁棒的 MAD

3.1 3-sigma 为什么在工业场景不够用

异常值检测最经典的方法是 3-sigma：算出均值和标准差，落在 ±3σ 之外的判为异常。但工业时序数据有两个特点会让 3-sigma 失效：

异常值本身会污染均值和标准差。一个几百倍的尖刺会把均值拉高、标准差拉大，导致本该被判异常的点反而落在 3σ 之内（"masking 效应"）。数据分布常常偏斜，不是对称的正态分布，对称的 ±3σ 区间不适用。

3.2 用 MAD 做鲁棒离群点检测

更稳健的做法是用 MAD（Median Absolute Deviation，中位数绝对偏差）。中位数对极端值不敏感，所以基于 MAD 的判断不会被尖刺带偏。

代码语言：SQL

复制

// 第一步：按设备算中位数和 MADrobustStats = selectmed(temperature)as medValue,med(abs(temperature - med(temperature)))as madValuefrom loadTable("dfs://iot", "temperature")group by deviceId// 第二步：算鲁棒 z-score，超过 3.5 判为离群点// 鲁棒 z-score = 0.6745 * (value - median) / MADoutliers = selectt.deviceId, t.ts, t.temperature,0.6745 * (t.temperature - r.medValue) r.madValue as robust_zfrom loadTable("dfs://iot", "temperature") as tinner join robustStats as r on t.deviceId = r.deviceIdwhere r.madValue > 0// MAD=0 时跳过，避免除零and abs(0.6745 * (t.temperature - r.medValue) r.madValue) > 3.5

阈值 3.5 是 MAD 方法常用的经验值（等价于正态分布下约 3-sigma，但对偏斜分布更鲁棒）。

img

3.3 物理约束：最可靠的一道闸

算完统计离群点后，还有一类异常统计方法抓不到——数值在统计上不异常，但物理上不可能。比如温度读出 -200°C、振动幅值为负、转速超过设备极限。

这类异常靠物理约束校验最直接，且不需要任何统计计算：

代码语言：SQL

复制

// 先做物理约束校验：只要超出合理区间，直接标记为 invalid invalid = select deviceId, ts, temperature from loadTable("dfs://iot", "temperature") where temperature < -50 or temperature > 200 // 温度物理范围 or isNull(temperature) // 空值同样视为无效 // 再把无效值统一置空，留给后续填充或直接忽略 update loadTable("dfs://iot", "temperature") set temperature = double(NULL) where temperature < -50 or temperature > 200

在工业场景里，物理约束往往是最划算的一类校验手段——不用额外计算统计量，也不用反复调阈值，直接把设备手册里的量程范围写进 where 条件，基本就能先拦下一批明显异常的数据。更稳妥的做法是，把它放在异常值处理的第一道关口，后面的统计方法再作为补充，这样效率和准确性通常都更均衡。

3.4 滑动窗口离群点：抓"相对突变"

前面几节抓的是"绝对异常"（相对全设备历史）。还有一种异常是"相对突变"——值在全局看不算异常，但相对设备自身最近一段时间的状态突然跳变。这种用滑动窗口抓：

代码语言：SQL

复制

// 滑动窗口离群点：当前值偏离近 60 秒均值超过 3 倍窗口标准差rel_outliers = select deviceId, ts, temperaturefrom (select deviceId, ts, temperature, ma vg(temperature, 60) as w_mean, mstd(temperature, 60) as w_stdfrom loadTable("dfs://iot", "temperature")context by deviceId)where w_std > 0and abs(temperature - w_mean) > 3 * w_std

这种检测对"工况切换时的真实跳变"和"传感器故障的尖刺"都会响应，需要结合工况标签（运行/停机）做二次过滤，避免把正常启停误判为异常。

四、卡死值：连续相同值的检测（IoT 特有）

这一节单独讲，因为卡死值是工业场景特有的、且最容易被漏检的脏数据。

4.1 卡死值为什么危险

传感器卡死时，上报的数据不是空、不是异常尖刺，而是连续若干个完全相同的值。从统计上看，这段数据"很平稳"，均值标准差都正常，但它完全不代表设备的真实状态。如果用这段数据算健康指标、做趋势分析，结论会系统性失真。

典型表现：一台本该持续波动的振动设备，连续 5 分钟上报完全相同的 RMS 值。

4.2 检测方法：连续相同值计数

检测思路：标记每个采样点是否与上一个相同，再统计连续相同的长度，超过阈值（比如连续 60 个采样点相同）判为卡死。

img

代码语言：SQL

复制

// 第一步：标记"是否与上一条相同"，并累计变化点形成"运行段"分组tagged = selectdeviceId, ts, value,(value = prev(value))as sameAsPrev, // 是否与上一条相同cumsum(iif(value <> prev(value), 1, 0))as runId // 变化点累加，形成段 IDfrom loadTable("dfs://iot", "vibration_rms")context by deviceId// 第二步：按段分组，统计每段长度，挑出异常长的"平稳段"stuck = selectdeviceId,min(ts)as stuck_start,max(ts)as stuck_end,count(*) as stuck_count,first(value) as stuck_valuefrom taggedgroup by deviceId, runIdha ving count(*) > 60// 连续超过 60 个相同值 = 卡死嫌疑

cumsum(iif(value <> prev(value), 1, 0)) 是个常用技巧：每次值变化时计数器加 1，相同则不变，于是同一段连续相同值共享同一个 runId。按 runId 分组后，count(*) 就是这段的长度。

4.3 处理策略：标记而非删除

检测到卡死值后，建议置空或打标记，而不是删除——因为卡死段的时间范围本身是有用的信息（什么时候开始卡、卡了多久），删掉就丢了故障溯源的线索。

代码语言：SQL

复制

// 给卡死段打标记，下游清洗时把这段值置空update loadTable("dfs://iot", "vibration_rms")set value = double(NULL)where (deviceId, ts) in (select deviceId, ts from taggedcontext by deviceId, runIdha ving count(*) > 60)

五、时序乱序：csort 与乱序监控

5.1 乱序从哪来

理想情况下，同一设备的数据按时间戳严格递增。但实际中，多源采集、网络延迟、时钟漂移都会让数据到达顺序和时间戳顺序不一致。如果上层直接按到达顺序做滑动窗口聚合，窗口边界就会错位。

5.2 排序：context by csort

读取数据时按设备分组、组内按时间排序，是处理乱序的标准做法：

代码语言：SQL

复制

// 读取时强制按设备分组、组内时间升序ordered = select *from loadTable("dfs://iot", "temperature")context by deviceIdcsort ts

context by deviceId 按设备分组，csort ts 在每个分组内按时间戳升序排列。这样下游的 prev、ma vg、deltas 等依赖时间顺序的函数才能正确工作。

5.3 乱序监控：别只排序，还要知道乱了多少

排序能修正顺序，但乱序严重本身是个信号——如果某台设备频繁乱序，往往说明采集链路或时钟有问题。把乱序程度量化出来，作为数据质量的一个监控指标：

代码语言：SQL

复制

// 统计每台设备的乱序点比例（时间戳小于前一条的点）disorder = selectdeviceId,count(*)as total,sum(iif(ts < prev(ts), 1, 0)) as disorder_count,sum(iif(ts < prev(ts), 1, 0)) count(*)as disorder_ratiofrom loadTable("dfs://iot", "temperature")context by deviceIdgroup by deviceId

disorder_ratio 高的设备，需要排查采集链路——可能是边缘网关缓冲不够、可能是 NTP 同步异常、也可能是多采集源合并时没去重。

六、把清洗固化成数据质量监控

前面四节是"发现问题"，这一节讲"持续盯住问题"。数据质量不是一次性清洗完就结束了，而是要定时跑、出报告、跟踪趋势——今天缺值率 1%，下周突然涨到 5%，说明上游出了状况，得有人知道。

img

6.1 构造数据质量指标（DQ scorecard）

给每类脏问题算一个比率指标，拼成一张质量记分卡：

代码语言：SQL

复制

// 按天、按设备算各类质量问题比率，形成质量记分卡dq = selectdate(ts)as day,deviceId,count(*)as total_rows,// 完整性：缺值率sum(isNull(temperature)) count(*) as missing_rate,// 准确性：物理越界率sum(iif(temperature < -50 or temperature > 200, 1, 0)) count(*)as invalid_rate,// 及时性：乱序率sum(iif(ts < prev(ts), 1, 0)) count(*)as disorder_rate,// 综合质量分 = 1 - 各类问题加权和1 - (sum(isNull(temperature)) sum(iif(temperature < -50 or temperature > 200, 1, 0))) 0 count(*) as quality_scorefrom loadTable("dfs://iot", "temperature")context by deviceIdgroup by date(ts), deviceId

quality_score 越接近 1 表示数据越干净。把它存到一张质量表里，就是数据质量的时间序列，本身也是个时序数据，可以用 DolphinDB 自己的看板和告警机制监控。

6.2 定时任务：每天自动跑

把上面的质量统计包成一个函数，用 scheduleJob 每天定时跑：

代码语言：SQL

复制

// 定义数据质量日报任务def dailyDQReport(day) {report = selectdate(ts) as day, deviceId,sum(isNull(temperature)) count(*) as missing_rate,sum(iif(temperature < -50 or temperature > 200, 1, 0)) count(*) as invalid_ratefrom loadTable("dfs://iot", "temperature")where date(ts) = daycontext by deviceIdgroup by date(ts), deviceId// 写入质量监控表loadTable("dfs://iot", "dq_report").tableInsert(report)// 质量差的设备直接告警（缺值率 > 10%）bad = select * from report where missing_rate > 0.1if(bad.size() > 0) {// 推送到告警流，由下游订阅处理loadTable("dfs://iot", "dq_alert_stream").tableInsert(bad)}}// 每天凌晨 1 点跑前一天的质量统计scheduleJob("daily_dq", "每日数据质量报告", dailyDQReport{date(now()) - 1},date(now()) 1, 01:00m, 'D')

这样数据质量就有了持续性：每天自动出报告，质量劣化的设备自动告警。运维不用被动等用户投诉"数据不对"，而是能主动发现上游问题。

七、写在最后

在工业 IoT 项目中，数据质量往往是最容易被忽视、却又最不能出问题的一环。很多团队把注意力放在流计算有多快、模型有多准、架构有多先进上，但别忘了，这一切其实都建立在“数据可信”这个前提之上。数据一旦不干净，实时预警就可能频繁误报，RUL 预测就会偏离实际，健康评分也会跟着失真。说到底，数据质量这件事不先解决，上层做得再炫，也不过是在沙地上起楼。

回到本文开头的区分：数据治理管生命周期，数据质量管正确性。两者是数据工程里平行的两件事——治理回答"数据该留多久、怎么瘦身"，质量回答"数据能不能信、哪里不可信"。一个完整的 IoT 数据平台，两层都要有。

落到 DolphinDB 上，数据质量处理的几个关键能力都齐备：

prev / deltas / ffill 处理缺值与填充med / mstd 做鲁棒统计和离群点检测context by csort / cumsum 处理乱序和连续段scheduleJob 把清洗固化成持续的质量监控

这些能力的关键不在单个函数多强大，而在它们都能用 SQL 在库内完成——检测、清洗、质量统计、定时监控，全程数据不用导出。对 IoT 这种数据量大的场景，少一次导出就少一次性能损耗和一致性问题。

最后留一个判断原则：清洗的目标不是把数据变"好看"，而是让下游知道数据有多可信。比起把缺值填满、把异常删掉，更重要的是把"哪里缺、哪里异常、哪里卡死"如实标记出来。诚实的不完整数据，比虚假的完整数据有价值得多。