做智能家居的人十有八九会在某个深夜盯着数据库里的设备上报记录发呆。一百个传感器每分钟上报一次一天跑出一千多万条数据早就超过了MySQL能舒服扛住的范围。实时告警压不下来历史统计跑又跑不动这时候你才意识到智能家居的本质不是几盏灯加一个开关而是一套7×24小时不停转的IoT数据管道。Lambda架构这套从推荐系统、搜索场景里历练出来的批流分离方案恰好能解决这堆问题。这篇文章就围绕我在智能家居大数据处理里的实际项目讲讲怎么用Lambda架构把设备数据从采集、清洗、实时计算到历史批处理完整串起来顺便把踩过的坑都摊开说。1. 为什么智能家居数据管道需要Lambda架构1.1 智能家居的数据规模没你想的那么小很多刚入行的朋友觉得家里几十个设备能有多大数据我以前也这么想直到我认真做了一次估算。一套普通的中户型智能家居常见的设备有温湿度传感器、人体存在传感器、门窗磁、灯光开关、空调伴侣、智能门锁、摄像头杂七杂八加起来大概30到50个。如果算上能耗监测和状态上报很多设备默认的上报频率是每5秒到30秒一次有的传感器甚至每秒都在报。我按最保守的30个设备、每10秒上报一次来算一天就是30×8640×24算下来单日消息量至少25万条。如果是100个设备、每5秒一次一天的记录数就是170万条以上。这还只是状态数据不算事件日志、告警记录、语音指令和摄像头产生的元数据。把这些全加起来一个家庭一个月轻轻松松产生几千万条记录物业管理下的整栋楼、整个小区数据量直接到亿级。这种规模下传统的关系型数据库基本撑不住。我试过用MySQL分表存前期跑得动但过了两三个月按月查询时索引膨胀、慢查询、锁表全来了。更麻烦的是业务方既要“当前家里温湿度是多少”这种毫秒级查询又要“过去30天能耗趋势”“某设备在线率”这类大规模聚合分析这两类需求放到一套系统里做怎么设计都会打架。1.2 实时与批量是两个必须同时满足的需求智能家居的数据处理天然分成两条路子。一条是实时路线门锁被异常开启要秒级告警燃气泄漏要立刻推送人回家要触发离家/回家模式这些场景要求数据从设备到用户手机端到端延迟最好控制在1到3秒以内。另一条是批量路线月底生成能耗账单、统计各房间平均温度、分析设备故障趋势、训练设备行为模型这些不追求实时但要求数据完整、口径统一、结果可回溯。关键问题来了同一份数据既要走秒级实时计算又要走全量离线统计怎么让两条链路不冲突、不重复、最终结果还能对得上我最早试过只做实时用Flink把窗口聚合结果写进数据库。可一旦要改计算口径或者想把某个设备过去一年的数据全量重算一遍Flink就得从Kafka里回放耗时长不说Kafka消息堆积也会把资源吃光。反过来只做跑批所有查询都是T1用户在App上看到的“实时温度”就是昨天的体验完全不能接受。1.3 Lambda架构的核心思想批流分离Lambda架构解决这个问题的方式很直白把数据通路拆成三层。批处理层负责全量、准确的计算不管数据是昨天还是去年的都能重新算产出的是可靠的基准数据速度层负责毫秒到秒级的实时计算只关心当前这几个窗口的数据弥补批处理层的延迟服务层把两边的结果合并起来对外提供统一查询。这个思路我后来跟朋友形容就像开一家饭店批处理层是后厨的总账本每天打烊后把全天流水核算清楚速度层是前台的点单系统保证客人坐下5分钟内就能吃上菜服务层就是菜单你要看常点的招牌菜还是今日时蔬它都给你端上来。这套架构选型还有一个很实际的好处两条链路互不干扰。实时链路挂了最多影响告警和实时状态批处理链路挂了历史查询还能用之前的快照顶住。对智能家居这种对稳定性要求很高的场景这个冗余的代价是值得的。当然也不是所有项目都适合Lambda如果数据量一天不到十万条实时需求也弱那直接MySQL加Redis就够了没必要杀鸡用牛刀。2. 整体架构设计与技术选型2.1 数据链路全景整个项目我用了这样一个结构设备端的采集工作由STM32网关和Home AssistantHA共同承担。STM32这类单片机负责接温湿度、人体红外、烟雾等传感器通过MQTT协议上报HA作为开源智能家居中枢把门锁、窗帘、灯、空调这些生态设备统一接入再通过它的API或MQTT桥接把数据转发出来。所有MQTT消息最终汇聚到EMQX再由一个轻量的数据桥接服务消费后写入Kafka完成最基础的数据缓冲和削峰。Kafka之后数据真正分流。速度层由Flink消费Kafka里的实时消息做事件时间窗口聚合把最近5分钟的平均温湿度、设备在线状态、异常事件直接写进Redis和HBase批处理层由一个Spring定时调度框架每天凌晨触发Spark作业读取Kafka落盘到HDFS的当日全量数据跑整套离线计算产出预聚合指标写回HBase和Elasticsearch。服务层是一组REST API查询时先读速度层的实时结果再按需回源批处理层的离线结果在内存里做合并。2.2 各层核心组件的选型分析这套架构里选型很关键用错一个组件整条链路都会别扭。我把核心组件的选型对比列在下面并说说当时为什么这么定。表格智能家居Lambda架构关键组件选型组件选型主要竞争者选型理由设备接入BrokerEMQXMosquitto、VerneMQ百万级连接能力、内置规则引擎插件丰富社区活跃消息缓冲Kafka 3.xPulsar、RabbitMQ生态成熟Flink/Spark集成度最高重放能力强实时计算引擎Flink 1.17Spark Streaming、Storm原生支持事件时间和水印状态管理完善适合IoT乱序数据批处理引擎Spark 3.xHive、Presto批处理性能稳内存计算快与HDFS/YARN配合完善实时结果存储Redis 7.0无毫秒级读写保存最近N个窗口的聚合值和设备状态明细存储HBase 2.xClickHouse、Doris、MongoDB天然支持海量时间序列写入rowkey按时间有序适合设备数据检索/分析Elasticsearch 8.xOpenSearch按设备、时间范围、类型做多维搜索页面报表靠它出图这里插一句为什么让MQTT进Kafka而不是让设备直接写Kafka因为设备端跑的是轻量级MQTT协议穿墙、断线重连、离线消息都有成熟机制而Kafka的客户端更重不适合跑在STM32和网关这种资源受限的设备上。通过EMQX做一次协议转换和消息过滤还能挡住不少非法设备和脏数据等于给后面的数据管道加了一层防火墙。2.3 为什么不做Kappa架构写这篇文章之前我知道一定会有人问Lambda又要写实时又要写批处理代码维护成本高为什么不用Kappa架构一条流全搞定我的回答是Kappa架构适合数据分析链路简单、重算需求少的场景但智能家居不是。智能家居的离线计算不只是“跑一遍出结果”还牵扯到大量维度的递归聚合。比如我要算某个设备在线率先按小时聚合再按天聚合最后按月生成报表。如果全在Flink里做每改一次口径就得保留从最早时间点开始的全部Kafka数据消息保留时间会拖到一两个月。更别提HA系统里设备经常被用户改名、换房间、换类型这类维度变更在流里很难优雅处理在批处理里一张离线表就能搞定。所以最终还是选了Lambda用Spark做离线维表、全量重算、数据订正用Flink做实时告警和实时状态展示两边各干各的反而省心。3. 核心实现从设备数据采集到服务层查询3.1 设备端接入与数据规整不管设备是走STM32直连还是接入HA第一步都是定一套统一的消息格式。我见过太多项目前期图省事温度上报字段写成temperature、temp、Temperature三种没有单位还混着摄氏度和华氏度数据分析的时候想死的心都有。我们的做法是统一JSON格式核心字段固定为device_id、event_type、ts、value、unit、extra其中ts是设备事件发生时的Unix毫秒时间戳value统一用字符串通过unit字段区分温度和湿度等不同含义。设备端STMP32上电后会先做SNTP时间同步这样上报的ts才可信。为什么强调这个因为后面Flink窗口计算如果拿设备本地时间当事件时间有的设备时钟偏了十分钟窗口聚合结果就是乱的。HA侧接入稍微特殊一点它不是直接上报原始数据而是把设备状态变化通过HA的事件总线触发再由一个自定义集成组件把数据转成标准JSON后发到EMQX的home/room/device主题。我实测下来这套方式比HA自带的recorder历史记录插件更适合大数据场景因为HA的SQLite存储一旦消息量大了读写会互相干扰。Kafka里的topic在设计上做了一个粗粒度的分区策略。消息key用device_id让同一设备的数据永远进同一个分区这样Flink按设备维度做窗口聚合时状态不用跨分区混洗。实测一个家庭150个设备Kafka用3个分区就够了但考虑后续接入更多住户我直接开了12个分区省得以后扩容时要重新分流。3.2 批处理层的落地实现批处理层我每天凌晨1点用Spring定时任务触发一个Spark作业处理的是前一天零点到24点的HDFS数据。Spark作业的主要逻辑分三块清洗、聚合、写结果。清洗阶段去掉value为空的记录、过滤掉明显越界的异常值比如温度-127或湿度大于100的传感器误报聚合阶段按device_id 小时窗口计算平均温度、最大最小温度、设备在线时长写结果阶段把聚合结果写进HBase的report表同时把明细数据写入Elasticsearch。下面给一段简化版的Spark聚合示意代码方便你理解结构。val df spark.read.parquet(hdfs:///iot/raw/2024/06/15) val hourAgg df .filter(col(value).isNotNull col(value) ! ) .withColumn(hour, from_unixtime(col(ts) / 1000, yyyy-MM-dd HH:00:00)) .groupBy(device_id, hour) .agg( avg(value).as(avg_value), min(value).as(min_value), max(value).as(max_value), count(value).as(sample_count) ) hourAgg.write .mode(overwrite) .option(zk, hbase-zookeeper:2181) .format(org.apache.hadoop.hbase.spark) .save(report:hour_agg)这版逻辑最大的坑在于如果数据源里混入了一些垃圾数据Spark离线作业的耗时和资源会成倍上升。我后来在清洗阶段加了白名单和值域校验把明显异常的数据单独写到error分区查询时能看到“脏数据量”指标排查问题方便很多。批处理层的产出并不是用来承担实时查询的它的核心作用是给服务层提供“昨天的准确结果”作为长期趋势的基准数据。3.3 速度层的实时链路速度层用的是Flink SQL因为代码量小、好维护。我建了一个source表连接Kafka的sensor_raw主题建了一个sink表连接Redis和HBase然后用一段连续SQL做5分钟的滚动窗口聚合。Flink作业部署在YARN上checkpoint间隔设置成20秒状态后端用RocksDB这样即使容器重启也不会丢状态。下面是当时用的Flink SQL片段去掉了业务细节保留了核心结构。CREATE TABLE sensor_source ( device_id STRING, event_type STRING, ts BIGINT, value STRING, status STRING, event_time AS TO_TIMESTAMP_LTZ(ts, 3), WATERMARK FOR event_time AS event_time - INTERVAL 30 SECOND ) WITH ( connector kafka, topic sensor_raw, properties.bootstrap.servers kafka01:9092, format json ); CREATE TABLE redis_sink ( device_id STRING, value DOUBLE, ts BIGINT ) WITH ( connector redis, redis.mode cluster, sink.key-pattern realtime:device_status ); INSERT INTO redis_sink SELECT device_id, CAST(value AS DOUBLE), MAX(ts) FROM sensor_source WHERE status online AND event_type telemetry GROUP BY device_id, TUMBLE(event_time, INTERVAL 10 SECOND);这段SQL干了什么它每10秒计算一次每个设备的最新状态值写入Redis这样App端打开首页能直接读到“客厅温度26.5℃”。真正的告警逻辑我没有放在Flink里而是让Flink只负责把设备异常事件比如燃气浓度超过阈值、门锁连续密码错误原样丢给一个轻量的规则引擎服务规则引擎再通过WebSocket推送给App端。为什么这么拆因为告警规则会频繁调整放在流计算里每次改都要重启作业放到规则引擎里改配置就生效运维成本更低。速度层还有一个重要任务就是给批处理层的计算结果提供中间态。比如在线率统计Flink每5分钟把设备在线时长累加到HBase的counter表到了凌晨Spark跑批时再基于这些累计值生成最终结果。这样做的好处是离线任务就算失败实时的累计值也不会丢。3.4 服务层合并查询设计服务层用Spring Boot提供查询接口核心逻辑就是“先快后准”。比如用户查看某个房间的温度曲线接口会先查Redis拿到最近2小时的实时聚合结果再查Elasticsearch拿到历史小时聚合结果两者在内存里拼接后返回。头两分钟的实时数据可能和最终批处理结果有几度的偏差但到了下一天Spark跑完离线任务后会把HBase里的report表更新成准确值用户刷新页面看到的就是修正后的曲线。为了保证修正确实生效我设计了一个version字段实时结果写入时version为“realtime:timestamp”批处理结果写入时version为“batch:yyyy-MM-dd”。查询时优先返回批处理结果批处理结果里没有覆盖到的最新时间段才用实时结果补位。这样处理虽然多写了一点代码但用户看到的数据永远朝“最终准确”靠拢而不是两头乱跳。4. 实操中踩过的坑与排查技巧4.1 数据乱序与设备时钟漂移第一个大坑就是乱序。STM32网关那边的传感器通过不同协议、不同中继节点上报到Kafka时顺序已经乱了。最崩溃的是有一批温湿度传感器时钟走得不准设备上报时间比真实时间慢了一个多小时。Flink窗口如果按Processing Time处理这几个传感器的数据会全部落进错误的窗口聚合出来的平均值完全不能用。解决办法是双管齐下。设备端加了SNTP校时每天凌晨自动校准一次Flink这边改用事件时间设了30秒的watermark延迟让乱序数据有足够时间到达。我还加了一个兜底逻辑如果某个事件的事件时间比当前时间早超过10分钟就把它打到单独侧输出流存进Kafka的late_data主题做后续分析不让它污染主链路。说实话前一周我几乎天天盯着这个侧输出流看后来数据量下降了才确认乱序问题被压住了。4.2 批流结果对不上做Lambda架构最头疼的问题就是白天看实时曲线是26.5℃第二天批处理跑完再看历史曲线变成了26.8℃。用户不一定会发现但做数据的人心里过不去。这背后的原因一般有三个一是时间窗口口径不一致Flink按事件时间窗口Spark按服务器本地时间分组两边切分点不同二是设备重启或上报重复实时链路没有去重批处理链路做了去重三是value类型不规范有的瞬间值被记录成字符串转成Double时精度丢失。我的解决办法是定义了一套统一的时间分桶规范把窗口边界统一到整5分钟和整小时上Flink和Spark都用同一个切分函数生成window_id。同时所有数据进入Kafka之前由桥接服务统一做一次幂等去重按device_id ts作为唯一键。这样一来两边算的是同一批数据结果自然就对齐了。这个经验和阿里那套“实时数仓和离线数仓对账”的思路很像核心就是口径统一。4.3 Flink背压与Spark资源抢占项目跑到第三个月设备数翻了一倍Flink作业开始出现背压警告Kafka消费延迟从几十秒慢慢涨到十几分钟。排查的时候先看的是Kafka消费速率确认生产端没瓶颈后才发现瓶颈在HBase的写入。设备数据高峰集中在晚上7点到11点大量写入涌向HBase的同一个region导致单点写入瓶颈和region分裂频繁。优化做了三件事。一是HBase的rowkey设计改成“倒序设备ID 小时前缀”同一小时内的数据尽可能散落到不同region二是Flink的sink端开启了批量写攒够200条或200毫秒再批量提交一次大幅减少网络往返三是把Spark批处理作业的时间从凌晨1点改到凌晨2点避开实时链路的晚高峰同时给Spark作业单独限制CPU和内存配额不让它抢完Flink的资源。改完以后背压问题基本消失消费延迟稳定在3秒以内。4.4 常见问题速查表我把整个项目里遇到的高频问题整理成了一张表方便其他人遇到同类情况时快速定位。表格智能家居Lambda架构常见问题排查速查表问题可能原因解决办法Kafka消费延迟持续增大HBase写入慢、region热点、sink线程少调整rowkey散列策略开启批量写增加sink并发Flink窗口数据漂移设备时钟不准、未用事件时间设备端SNTP校时Flink使用事件时间并设置watermark实时与离线结果不一致窗口口径不一致、未去重统一window_id切分规则进入Kafka前按device_idts去重Spark跑批时间越来越长脏数据太多、文件小文件过多清洗阶段过滤异常值定期对HDFS小文件做合并Redis缓存雪崩所有key同时过期过期时间加随机化设置永不过期并靠定时任务刷新HBase热点写入rowkey顺序单调递增加盐或倒序处理rowkey前缀HA系统历史记录拖慢操作SQLite存储读写互相干扰只保留关键事件原始数据全部转发到Kafka由大数据管道存储另外补充一个很多人忽略的小经验Kafka的topic分区数最好一次性规划到位后期扩容分区会导致同一key的数据跑到不同分区影响排序和窗口聚合。我当时就是因为一开始只建了3个分区后来不得不加结果花了整整两天处理乱序和重复消费的问题。从第一天就按未来半年设备增长量规划好分区数能省掉后面所有的麻烦。我个人在实际操作中最大的体会是Lambda架构在智能家居这个场景里不是一道理论题而是一套非常务实的基础设施。它结构上比单流处理复杂一点但换来的却是实时响应和离线统计两不耽误而且故障隔离效果很好实时链路挂了不影响历史数据批处理链路挂了也不影响当下的告警。这个项目跑了大半年上线初期我每天都在处理上面这些琐碎的问题但等链路稳定之后整个数据平台就变得特别“安静”几乎不用人干预。最后再分享一个小建议不要一上来就照着架构图把所有组件全部铺开。先拿一台设备的数据搭一条最简陋的Flink到Redis的实时链路确认端到端能跑通再补上Spark的离线批处理和HBase存储最后再慢慢加HA接入、Elasticsearch、告警规则这些周边能力。每加一层验证一层出问题时定位范围就小得多。Lambda架构本身不复杂复杂的是设备永远比你预想的更不可控。
智能家居大数据管道：Lambda架构批流分离实战解析
做智能家居的人十有八九会在某个深夜盯着数据库里的设备上报记录发呆。一百个传感器每分钟上报一次一天跑出一千多万条数据早就超过了MySQL能舒服扛住的范围。实时告警压不下来历史统计跑又跑不动这时候你才意识到智能家居的本质不是几盏灯加一个开关而是一套7×24小时不停转的IoT数据管道。Lambda架构这套从推荐系统、搜索场景里历练出来的批流分离方案恰好能解决这堆问题。这篇文章就围绕我在智能家居大数据处理里的实际项目讲讲怎么用Lambda架构把设备数据从采集、清洗、实时计算到历史批处理完整串起来顺便把踩过的坑都摊开说。1. 为什么智能家居数据管道需要Lambda架构1.1 智能家居的数据规模没你想的那么小很多刚入行的朋友觉得家里几十个设备能有多大数据我以前也这么想直到我认真做了一次估算。一套普通的中户型智能家居常见的设备有温湿度传感器、人体存在传感器、门窗磁、灯光开关、空调伴侣、智能门锁、摄像头杂七杂八加起来大概30到50个。如果算上能耗监测和状态上报很多设备默认的上报频率是每5秒到30秒一次有的传感器甚至每秒都在报。我按最保守的30个设备、每10秒上报一次来算一天就是30×8640×24算下来单日消息量至少25万条。如果是100个设备、每5秒一次一天的记录数就是170万条以上。这还只是状态数据不算事件日志、告警记录、语音指令和摄像头产生的元数据。把这些全加起来一个家庭一个月轻轻松松产生几千万条记录物业管理下的整栋楼、整个小区数据量直接到亿级。这种规模下传统的关系型数据库基本撑不住。我试过用MySQL分表存前期跑得动但过了两三个月按月查询时索引膨胀、慢查询、锁表全来了。更麻烦的是业务方既要“当前家里温湿度是多少”这种毫秒级查询又要“过去30天能耗趋势”“某设备在线率”这类大规模聚合分析这两类需求放到一套系统里做怎么设计都会打架。1.2 实时与批量是两个必须同时满足的需求智能家居的数据处理天然分成两条路子。一条是实时路线门锁被异常开启要秒级告警燃气泄漏要立刻推送人回家要触发离家/回家模式这些场景要求数据从设备到用户手机端到端延迟最好控制在1到3秒以内。另一条是批量路线月底生成能耗账单、统计各房间平均温度、分析设备故障趋势、训练设备行为模型这些不追求实时但要求数据完整、口径统一、结果可回溯。关键问题来了同一份数据既要走秒级实时计算又要走全量离线统计怎么让两条链路不冲突、不重复、最终结果还能对得上我最早试过只做实时用Flink把窗口聚合结果写进数据库。可一旦要改计算口径或者想把某个设备过去一年的数据全量重算一遍Flink就得从Kafka里回放耗时长不说Kafka消息堆积也会把资源吃光。反过来只做跑批所有查询都是T1用户在App上看到的“实时温度”就是昨天的体验完全不能接受。1.3 Lambda架构的核心思想批流分离Lambda架构解决这个问题的方式很直白把数据通路拆成三层。批处理层负责全量、准确的计算不管数据是昨天还是去年的都能重新算产出的是可靠的基准数据速度层负责毫秒到秒级的实时计算只关心当前这几个窗口的数据弥补批处理层的延迟服务层把两边的结果合并起来对外提供统一查询。这个思路我后来跟朋友形容就像开一家饭店批处理层是后厨的总账本每天打烊后把全天流水核算清楚速度层是前台的点单系统保证客人坐下5分钟内就能吃上菜服务层就是菜单你要看常点的招牌菜还是今日时蔬它都给你端上来。这套架构选型还有一个很实际的好处两条链路互不干扰。实时链路挂了最多影响告警和实时状态批处理链路挂了历史查询还能用之前的快照顶住。对智能家居这种对稳定性要求很高的场景这个冗余的代价是值得的。当然也不是所有项目都适合Lambda如果数据量一天不到十万条实时需求也弱那直接MySQL加Redis就够了没必要杀鸡用牛刀。2. 整体架构设计与技术选型2.1 数据链路全景整个项目我用了这样一个结构设备端的采集工作由STM32网关和Home AssistantHA共同承担。STM32这类单片机负责接温湿度、人体红外、烟雾等传感器通过MQTT协议上报HA作为开源智能家居中枢把门锁、窗帘、灯、空调这些生态设备统一接入再通过它的API或MQTT桥接把数据转发出来。所有MQTT消息最终汇聚到EMQX再由一个轻量的数据桥接服务消费后写入Kafka完成最基础的数据缓冲和削峰。Kafka之后数据真正分流。速度层由Flink消费Kafka里的实时消息做事件时间窗口聚合把最近5分钟的平均温湿度、设备在线状态、异常事件直接写进Redis和HBase批处理层由一个Spring定时调度框架每天凌晨触发Spark作业读取Kafka落盘到HDFS的当日全量数据跑整套离线计算产出预聚合指标写回HBase和Elasticsearch。服务层是一组REST API查询时先读速度层的实时结果再按需回源批处理层的离线结果在内存里做合并。2.2 各层核心组件的选型分析这套架构里选型很关键用错一个组件整条链路都会别扭。我把核心组件的选型对比列在下面并说说当时为什么这么定。表格智能家居Lambda架构关键组件选型组件选型主要竞争者选型理由设备接入BrokerEMQXMosquitto、VerneMQ百万级连接能力、内置规则引擎插件丰富社区活跃消息缓冲Kafka 3.xPulsar、RabbitMQ生态成熟Flink/Spark集成度最高重放能力强实时计算引擎Flink 1.17Spark Streaming、Storm原生支持事件时间和水印状态管理完善适合IoT乱序数据批处理引擎Spark 3.xHive、Presto批处理性能稳内存计算快与HDFS/YARN配合完善实时结果存储Redis 7.0无毫秒级读写保存最近N个窗口的聚合值和设备状态明细存储HBase 2.xClickHouse、Doris、MongoDB天然支持海量时间序列写入rowkey按时间有序适合设备数据检索/分析Elasticsearch 8.xOpenSearch按设备、时间范围、类型做多维搜索页面报表靠它出图这里插一句为什么让MQTT进Kafka而不是让设备直接写Kafka因为设备端跑的是轻量级MQTT协议穿墙、断线重连、离线消息都有成熟机制而Kafka的客户端更重不适合跑在STM32和网关这种资源受限的设备上。通过EMQX做一次协议转换和消息过滤还能挡住不少非法设备和脏数据等于给后面的数据管道加了一层防火墙。2.3 为什么不做Kappa架构写这篇文章之前我知道一定会有人问Lambda又要写实时又要写批处理代码维护成本高为什么不用Kappa架构一条流全搞定我的回答是Kappa架构适合数据分析链路简单、重算需求少的场景但智能家居不是。智能家居的离线计算不只是“跑一遍出结果”还牵扯到大量维度的递归聚合。比如我要算某个设备在线率先按小时聚合再按天聚合最后按月生成报表。如果全在Flink里做每改一次口径就得保留从最早时间点开始的全部Kafka数据消息保留时间会拖到一两个月。更别提HA系统里设备经常被用户改名、换房间、换类型这类维度变更在流里很难优雅处理在批处理里一张离线表就能搞定。所以最终还是选了Lambda用Spark做离线维表、全量重算、数据订正用Flink做实时告警和实时状态展示两边各干各的反而省心。3. 核心实现从设备数据采集到服务层查询3.1 设备端接入与数据规整不管设备是走STM32直连还是接入HA第一步都是定一套统一的消息格式。我见过太多项目前期图省事温度上报字段写成temperature、temp、Temperature三种没有单位还混着摄氏度和华氏度数据分析的时候想死的心都有。我们的做法是统一JSON格式核心字段固定为device_id、event_type、ts、value、unit、extra其中ts是设备事件发生时的Unix毫秒时间戳value统一用字符串通过unit字段区分温度和湿度等不同含义。设备端STMP32上电后会先做SNTP时间同步这样上报的ts才可信。为什么强调这个因为后面Flink窗口计算如果拿设备本地时间当事件时间有的设备时钟偏了十分钟窗口聚合结果就是乱的。HA侧接入稍微特殊一点它不是直接上报原始数据而是把设备状态变化通过HA的事件总线触发再由一个自定义集成组件把数据转成标准JSON后发到EMQX的home/room/device主题。我实测下来这套方式比HA自带的recorder历史记录插件更适合大数据场景因为HA的SQLite存储一旦消息量大了读写会互相干扰。Kafka里的topic在设计上做了一个粗粒度的分区策略。消息key用device_id让同一设备的数据永远进同一个分区这样Flink按设备维度做窗口聚合时状态不用跨分区混洗。实测一个家庭150个设备Kafka用3个分区就够了但考虑后续接入更多住户我直接开了12个分区省得以后扩容时要重新分流。3.2 批处理层的落地实现批处理层我每天凌晨1点用Spring定时任务触发一个Spark作业处理的是前一天零点到24点的HDFS数据。Spark作业的主要逻辑分三块清洗、聚合、写结果。清洗阶段去掉value为空的记录、过滤掉明显越界的异常值比如温度-127或湿度大于100的传感器误报聚合阶段按device_id 小时窗口计算平均温度、最大最小温度、设备在线时长写结果阶段把聚合结果写进HBase的report表同时把明细数据写入Elasticsearch。下面给一段简化版的Spark聚合示意代码方便你理解结构。val df spark.read.parquet(hdfs:///iot/raw/2024/06/15) val hourAgg df .filter(col(value).isNotNull col(value) ! ) .withColumn(hour, from_unixtime(col(ts) / 1000, yyyy-MM-dd HH:00:00)) .groupBy(device_id, hour) .agg( avg(value).as(avg_value), min(value).as(min_value), max(value).as(max_value), count(value).as(sample_count) ) hourAgg.write .mode(overwrite) .option(zk, hbase-zookeeper:2181) .format(org.apache.hadoop.hbase.spark) .save(report:hour_agg)这版逻辑最大的坑在于如果数据源里混入了一些垃圾数据Spark离线作业的耗时和资源会成倍上升。我后来在清洗阶段加了白名单和值域校验把明显异常的数据单独写到error分区查询时能看到“脏数据量”指标排查问题方便很多。批处理层的产出并不是用来承担实时查询的它的核心作用是给服务层提供“昨天的准确结果”作为长期趋势的基准数据。3.3 速度层的实时链路速度层用的是Flink SQL因为代码量小、好维护。我建了一个source表连接Kafka的sensor_raw主题建了一个sink表连接Redis和HBase然后用一段连续SQL做5分钟的滚动窗口聚合。Flink作业部署在YARN上checkpoint间隔设置成20秒状态后端用RocksDB这样即使容器重启也不会丢状态。下面是当时用的Flink SQL片段去掉了业务细节保留了核心结构。CREATE TABLE sensor_source ( device_id STRING, event_type STRING, ts BIGINT, value STRING, status STRING, event_time AS TO_TIMESTAMP_LTZ(ts, 3), WATERMARK FOR event_time AS event_time - INTERVAL 30 SECOND ) WITH ( connector kafka, topic sensor_raw, properties.bootstrap.servers kafka01:9092, format json ); CREATE TABLE redis_sink ( device_id STRING, value DOUBLE, ts BIGINT ) WITH ( connector redis, redis.mode cluster, sink.key-pattern realtime:device_status ); INSERT INTO redis_sink SELECT device_id, CAST(value AS DOUBLE), MAX(ts) FROM sensor_source WHERE status online AND event_type telemetry GROUP BY device_id, TUMBLE(event_time, INTERVAL 10 SECOND);这段SQL干了什么它每10秒计算一次每个设备的最新状态值写入Redis这样App端打开首页能直接读到“客厅温度26.5℃”。真正的告警逻辑我没有放在Flink里而是让Flink只负责把设备异常事件比如燃气浓度超过阈值、门锁连续密码错误原样丢给一个轻量的规则引擎服务规则引擎再通过WebSocket推送给App端。为什么这么拆因为告警规则会频繁调整放在流计算里每次改都要重启作业放到规则引擎里改配置就生效运维成本更低。速度层还有一个重要任务就是给批处理层的计算结果提供中间态。比如在线率统计Flink每5分钟把设备在线时长累加到HBase的counter表到了凌晨Spark跑批时再基于这些累计值生成最终结果。这样做的好处是离线任务就算失败实时的累计值也不会丢。3.4 服务层合并查询设计服务层用Spring Boot提供查询接口核心逻辑就是“先快后准”。比如用户查看某个房间的温度曲线接口会先查Redis拿到最近2小时的实时聚合结果再查Elasticsearch拿到历史小时聚合结果两者在内存里拼接后返回。头两分钟的实时数据可能和最终批处理结果有几度的偏差但到了下一天Spark跑完离线任务后会把HBase里的report表更新成准确值用户刷新页面看到的就是修正后的曲线。为了保证修正确实生效我设计了一个version字段实时结果写入时version为“realtime:timestamp”批处理结果写入时version为“batch:yyyy-MM-dd”。查询时优先返回批处理结果批处理结果里没有覆盖到的最新时间段才用实时结果补位。这样处理虽然多写了一点代码但用户看到的数据永远朝“最终准确”靠拢而不是两头乱跳。4. 实操中踩过的坑与排查技巧4.1 数据乱序与设备时钟漂移第一个大坑就是乱序。STM32网关那边的传感器通过不同协议、不同中继节点上报到Kafka时顺序已经乱了。最崩溃的是有一批温湿度传感器时钟走得不准设备上报时间比真实时间慢了一个多小时。Flink窗口如果按Processing Time处理这几个传感器的数据会全部落进错误的窗口聚合出来的平均值完全不能用。解决办法是双管齐下。设备端加了SNTP校时每天凌晨自动校准一次Flink这边改用事件时间设了30秒的watermark延迟让乱序数据有足够时间到达。我还加了一个兜底逻辑如果某个事件的事件时间比当前时间早超过10分钟就把它打到单独侧输出流存进Kafka的late_data主题做后续分析不让它污染主链路。说实话前一周我几乎天天盯着这个侧输出流看后来数据量下降了才确认乱序问题被压住了。4.2 批流结果对不上做Lambda架构最头疼的问题就是白天看实时曲线是26.5℃第二天批处理跑完再看历史曲线变成了26.8℃。用户不一定会发现但做数据的人心里过不去。这背后的原因一般有三个一是时间窗口口径不一致Flink按事件时间窗口Spark按服务器本地时间分组两边切分点不同二是设备重启或上报重复实时链路没有去重批处理链路做了去重三是value类型不规范有的瞬间值被记录成字符串转成Double时精度丢失。我的解决办法是定义了一套统一的时间分桶规范把窗口边界统一到整5分钟和整小时上Flink和Spark都用同一个切分函数生成window_id。同时所有数据进入Kafka之前由桥接服务统一做一次幂等去重按device_id ts作为唯一键。这样一来两边算的是同一批数据结果自然就对齐了。这个经验和阿里那套“实时数仓和离线数仓对账”的思路很像核心就是口径统一。4.3 Flink背压与Spark资源抢占项目跑到第三个月设备数翻了一倍Flink作业开始出现背压警告Kafka消费延迟从几十秒慢慢涨到十几分钟。排查的时候先看的是Kafka消费速率确认生产端没瓶颈后才发现瓶颈在HBase的写入。设备数据高峰集中在晚上7点到11点大量写入涌向HBase的同一个region导致单点写入瓶颈和region分裂频繁。优化做了三件事。一是HBase的rowkey设计改成“倒序设备ID 小时前缀”同一小时内的数据尽可能散落到不同region二是Flink的sink端开启了批量写攒够200条或200毫秒再批量提交一次大幅减少网络往返三是把Spark批处理作业的时间从凌晨1点改到凌晨2点避开实时链路的晚高峰同时给Spark作业单独限制CPU和内存配额不让它抢完Flink的资源。改完以后背压问题基本消失消费延迟稳定在3秒以内。4.4 常见问题速查表我把整个项目里遇到的高频问题整理成了一张表方便其他人遇到同类情况时快速定位。表格智能家居Lambda架构常见问题排查速查表问题可能原因解决办法Kafka消费延迟持续增大HBase写入慢、region热点、sink线程少调整rowkey散列策略开启批量写增加sink并发Flink窗口数据漂移设备时钟不准、未用事件时间设备端SNTP校时Flink使用事件时间并设置watermark实时与离线结果不一致窗口口径不一致、未去重统一window_id切分规则进入Kafka前按device_idts去重Spark跑批时间越来越长脏数据太多、文件小文件过多清洗阶段过滤异常值定期对HDFS小文件做合并Redis缓存雪崩所有key同时过期过期时间加随机化设置永不过期并靠定时任务刷新HBase热点写入rowkey顺序单调递增加盐或倒序处理rowkey前缀HA系统历史记录拖慢操作SQLite存储读写互相干扰只保留关键事件原始数据全部转发到Kafka由大数据管道存储另外补充一个很多人忽略的小经验Kafka的topic分区数最好一次性规划到位后期扩容分区会导致同一key的数据跑到不同分区影响排序和窗口聚合。我当时就是因为一开始只建了3个分区后来不得不加结果花了整整两天处理乱序和重复消费的问题。从第一天就按未来半年设备增长量规划好分区数能省掉后面所有的麻烦。我个人在实际操作中最大的体会是Lambda架构在智能家居这个场景里不是一道理论题而是一套非常务实的基础设施。它结构上比单流处理复杂一点但换来的却是实时响应和离线统计两不耽误而且故障隔离效果很好实时链路挂了不影响历史数据批处理链路挂了也不影响当下的告警。这个项目跑了大半年上线初期我每天都在处理上面这些琐碎的问题但等链路稳定之后整个数据平台就变得特别“安静”几乎不用人干预。最后再分享一个小建议不要一上来就照着架构图把所有组件全部铺开。先拿一台设备的数据搭一条最简陋的Flink到Redis的实时链路确认端到端能跑通再补上Spark的离线批处理和HBase存储最后再慢慢加HA接入、Elasticsearch、告警规则这些周边能力。每加一层验证一层出问题时定位范围就小得多。Lambda架构本身不复杂复杂的是设备永远比你预想的更不可控。