3、数据清洗与预处理：边缘端数据去重、异常值过滤、时间戳对齐
好，咱们接着聊。上一节我们把数据采集上来了，但说实话，从现场设备拿到的数据，那叫一个“脏”。
我见过最夸张的一次，一个温度传感器一天报了三千多次重复数据，还有的时间戳直接跳到了2038年。你想想看，这种数据要是直接送到云端做分析，那结果能看吗？
所以，数据清洗与预处理，就是咱们在边缘端要干的第一件正事。说白了，就是把垃圾扔出去，把有用的留下，再给它们排好队。
3.1 数据去重：别让重复数据浪费带宽
为什么会重复？原因很多。传感器误触发、网络重传、设备重启后重新上报……我在一个水处理项目中遇到过，一个液位计每5秒上报一次数据，但PLC那边因为通讯不稳定，经常把同一包数据重发了三四次。
去重的核心思路其实很简单：相同设备、相同时间、相同数值，只保留一条。
但在边缘端实现，咱们得考虑性能。我建议用滑动窗口+哈希表的方案：
// 伪代码示例：边缘端数据去重
class DataDeduplicator {
// 滑动窗口，保存最近N条数据的哈希值
private window: Set<string> = new Set()
private maxWindowSize: number = 1000
public isDuplicate(data: SensorData): boolean {
// 生成唯一键：设备ID + 时间戳(秒级) + 数值(取整)
const key = `${data.deviceId}_${Math.floor(data.timestamp/1000)}_${Math.round(data.value * 100)}`
if (this.window.has(key)) {
return true // 重复，丢弃
}
// 新数据，加入窗口
this.window.add(key)
// 控制窗口大小，防止内存溢出
if (this.window.size > this.maxWindowSize) {
// 移除最旧的数据（这里简化处理，实际可用LRU）
const firstKey = this.window.values().next().value
this.window.delete(firstKey)
}
return false
}
}
我的经验：窗口大小别设太大。在资源受限的边缘设备上，1000条足够了。我曾经试过设成10000，结果内存直接爆了。另外，时间戳对齐到秒级就够了，毫秒级去重反而容易误杀正常数据。
3.2 异常值过滤：把“疯子”数据拦在门外
异常值，我管它叫“疯子数据”。明明是个室温传感器，突然报了个150度；明明电流最大100A，突然蹦出个9999A。这种数据一旦进入分析系统，轻则报警误报，重则导致整个控制逻辑崩溃。
常用的过滤方法有三种，我按推荐程度排个序：
| 方法 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 阈值过滤 | 所有场景（基础必备） | 简单、高效、零计算开销 | 需要人工设定上下限 |
| 3σ原则 | 数据分布较稳定的场景 | 自适应，无需人工设阈值 | 对突变不敏感，计算稍复杂 |
| 中位数绝对偏差 | 存在较多毛刺的场景 | 抗干扰能力强 | 需要缓存窗口数据 |
我个人习惯是阈值过滤打底，再叠加一种自适应方法。比如这样：
// 异常值过滤示例：阈值 + 3σ
class AnomalyFilter {
private thresholdMin: number
private thresholdMax: number
private historyValues: number[] = []
private readonly historySize = 100
constructor(min: number, max: number) {
this.thresholdMin = min
this.thresholdMax = max
}
public isAnomaly(value: number): boolean {
// 第一步：硬阈值过滤
if (value < this.thresholdMin || value > this.thresholdMax) {
return true
}
// 第二步：3σ检测（需要积累一定数据量）
if (this.historyValues.length >= 30) {
const mean = this.historyValues.reduce((a, b) => a + b, 0) / this.historyValues.length
const variance = this.historyValues.reduce((sum, v) => sum + Math.pow(v - mean, 2), 0) / this.historyValues.length
const std = Math.sqrt(variance)
if (Math.abs(value - mean) > 3 * std) {
return true // 超出3σ范围，判为异常
}
}
// 更新历史窗口
this.historyValues.push(value)
if (this.historyValues.length > this.historySize) {
this.historyValues.shift()
}
return false
}
}
注意：3σ方法有个坑——如果历史数据本身就包含异常值，均值和标准差都会被带偏。我曾经在一个振动监测项目上吃过这个亏，连续三天没报异常，结果发现是历史窗口里混进了异常数据，把标准差撑大了。解决方案：先做一次硬阈值过滤，再进3σ计算。
3.3 时间戳对齐：让数据说同一种“时间语言”
这个问题在异构设备接入时特别突出。有的设备用UTC时间，有的用本地时间，还有的用设备开机后的相对时间。更头疼的是，不同设备的采样周期还不一样——一个1秒采一次，另一个3秒采一次。
时间戳对齐，说白了就是把所有数据的时间基准统一，然后按固定的时间窗口重新组织。
我常用的方法是线性插值对齐法：
// 时间戳对齐示例：线性插值
class TimeAligner {
private baseInterval: number = 1000 // 对齐到1秒间隔（毫秒）
// 输入：原始数据流，输出：对齐后的数据
public align(rawData: SensorData[]): AlignedData[] {
if (rawData.length < 2) return []
// 1. 统一转为UTC毫秒时间戳
const utcData = rawData.map(d => ({
...d,
timestamp: this.toUtcMs(d.timestamp, d.timezone)
}))
// 2. 找到时间范围
const startTime = Math.ceil(utcData[0].timestamp / this.baseInterval) * this.baseInterval
const endTime = Math.floor(utcData[utcData.length - 1].timestamp / this.baseInterval) * this.baseInterval
const result: AlignedData[] = []
// 3. 按固定间隔插值
for (let t = startTime; t <= endTime; t += this.baseInterval) {
// 找到t前后的两个数据点
const before = utcData.filter(d => d.timestamp <= t).pop()
const after = utcData.find(d => d.timestamp >= t)
if (before && after) {
// 线性插值
const ratio = (t - before.timestamp) / (after.timestamp - before.timestamp)
const interpolatedValue = before.value + (after.value - before.value) * ratio
result.push({
timestamp: t,
value: Math.round(interpolatedValue * 100) / 100,
deviceId: rawData[0].deviceId
})
}
}
return result
}
private toUtcMs(timestamp: any, timezone?: string): number {
// 实际项目中需要处理各种时间格式
// 这里简化处理
return Number(timestamp)
}
}
核心要点：时间戳对齐不是简单的“取整”，而是要考虑数据本身的物理意义。比如温度变化是连续的，用线性插值没问题；但开关量（0/1）就不能插值，得用“最近邻”法，取最接近时间点的值。
3.4 整体流程：一张图看懂
上面三个步骤，在实际项目中是串在一起的。我画了张流程图，帮你理清关系：
你看，数据从左边进来，先过“去重”关，再过“异常过滤”关，最后做“时间对齐”。三道关卡下来，数据质量就有保障了。
我的建议：这三步的顺序不要乱。先做去重，可以减少后续过滤和对齐的计算量。我曾经试过先做时间对齐再做去重，结果因为插值产生了大量“伪重复”数据，去重逻辑反而把有效数据误杀了。
3.5 避坑指南
最后，分享几个我踩过的坑：
- 别在边缘端做太复杂的清洗——边缘设备资源有限，CPU和内存都很金贵。把明显的脏数据去掉就行，复杂的统计分析留到云端做。
- 保留原始数据——清洗后的数据用于实时分析，但原始数据最好也存一份（哪怕只存24小时）。万一清洗逻辑出问题，还能回溯。
- 给清洗操作打标签——每条数据经过清洗后，建议加个字段记录它经历了哪些处理。比如
cleaning_flags: ["dedup", "threshold_filter", "interpolated"]。这样排查问题时一目了然。
嗯，数据清洗这块就聊到这儿。说白了，就是三个字：去、滤、对齐。把这三点做好了，后面的数据分析工作才能站得住脚。