一、问题场景一条产线每秒上千条传感器数据某电机产线每台设备有温度、振动、电流 3 路传感器采样频率 10Hz。100 台设备并发 →每秒约 3000 条遥测。传统做法是存下来再分析等发现轴承过热产线可能已经烧了。我们要的闭环是边缘采样 → Kafka 汇聚 → Flink 实时算异常 → 云端告警/看板 → 反向下发降速指令。本篇聚焦中间那段实时异常检测也是周五连载云边协同的大脑部分。二、方案设计整体数据流[边缘网关] --MQTT-- [Kafka topic: sensor.raw] | [Flink Job] keyBy(deviceId) → 滑动窗口(z-score) → 异常判定 → ├─ 正常 → 写入时序库(Put) └─ 异常 → 告警(WebSocket/邮件) 标记为什么用z-score 滑动窗口而不是简单阈值因为同一台电机在不同工况下正常温度不一样绝对阈值会误报。用最近窗口的均值/标准差做动态基线更鲁棒。三、分步实现PyFlink可读性优先1. 定义数据结构与源from pyflink.datastream import StreamExecutionEnvironment from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsets from pyflink.common.serialization import SimpleStringSchema from pyflink.common.watermark_strategy import WatermarkStrategy import json ​ env StreamExecutionEnvironment.get_execution_environment() env.set_parallelism(4) ​ source KafkaSource.builder() \ .set_bootstrap_servers(kafka:9092) \ .set_topics(sensor.raw) \ .set_group_id(flink-anomaly) \ .set_starting_offsets(KafkaOffsets.latest()) \ .set_value_only_deserializer(SimpleStringSchema()) \ .build() ​ ds env.from_source(source, WatermarkStrategy.no_watermarks(), kafka)2. 解析 keyBy 设备def parse(record): e json.loads(record) return (e[deviceId], e[metric], float(e[value]), int(e[ts])) ​ parsed ds.map(parse, output_type...) keyed parsed.key_by(lambda x: (x[0], x[1])) # 按 设备指标 分组3. 滑动窗口 z-score 异常检测核心算子from pyflink.datastream.window import SlidingEventTimeWindows from pyflink.common.time import Time ​ windowed keyed \ .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.seconds(30))) \ .process(AnomalyDetector()) class AnomalyDetector(KeyedProcessWindowFunction): def process(self, key, ctx, events): vals sorted([e[2] for e in events]) n len(vals) mean sum(vals) / n var sum((v - mean) ** 2 for v in vals) / n std var ** 0.5 1e-6 # 用窗口末尾的点做 z-score 判定 latest vals[-1] z (latest - mean) / std if abs(z) 3.0: # 3σ 准则 yield { deviceId: key[0], metric: key[1], value: latest, z: round(z, 2), mean: round(mean, 2), ts: ctx.current_watermark }4. 异常分流到告警 Sinkanomalies windowed.map(lambda a: json.dumps(a)) anomalies.add_sink(KafkaSink.builder() .set_bootstrap_servers(kafka:9092) .set_record_serializer(..., topicsensor.alert) .build())下游一个 Spring Boot / Node 服务订阅sensor.alert推 WebSocket 到运维看板并按设备 ID 触发降级指令回写边缘网关。四、踩坑记录乱序事件必须有 Watermark工业网关网络抖动事件迟到是常态。不设 watermark 允许的延迟窗口会提前触发导致漏检。状态膨胀keyBy(deviceId, metric)后窗口状态随时间增长务必配State TTL否则一周后 JobManager 内存爆炸。z-score 对突发不敏感纯统计方法抓不出缓变劣化。生产里常叠加斜率检测 / EWMA本篇留给进阶版。不要在 process 里查数据库每条事件去查设备元数据会拖垮吞吐预先广播BroadcastState下发设备配置。五、性能数据单机基准指标数值吞吐单 TaskManager4 核约12 万 events/s端到端延迟采样→告警p99 800ms100 台设备 3 路传感器稳态 CPU ~55%Flink 把事后看报表变成了事中拦风险。这套管道正是我们整个 Edge AI 全栈的数据主动脉——边缘负责采和跑轻模型云端 Flink 负责 aggregation 和全局异常判定。