简介这份源码资源面向物联网、大数据方向的开发者与学习者聚焦智能家居设备数据的采集、传输与分析全链路解决从传感器数据接入到实时可视化展示的完整实践问题。项目以MQTT协议收集设备数据借助Kafka消息队列保障实时性与可靠性通过HDFS完成大规模存储再由Spark进行高效处理分析并将结果写入PostgreSQL最终提供Web实时仪表板与NiFi可视化界面方便监控系统状态并开展初步分析。资源包共16个文件包含zip依赖库、db数据库文件、yml容器编排配置、sql建表脚本、conf与env环境配置、ino设备端程序、py处理脚本及sh启动脚本等压缩包约174KB结构紧凑、模块清晰。目前已有53人学习下载适合希望打通智能家居数据管道、理解Spark与Kafka协同工作方式的读者参考可据此快速搭建实验环境并复用数据处理与可视化思路。1. 从一份智能家居数据说起Spark 和 Kafka 到底在系统里干什么智能家居设备一旦上了规模数据就不再是「几条温湿度记录」那么简单。一个三居室全屋智能门磁、人体红外、温湿度、插座功率、摄像头事件、网关心跳加起来轻松上百个数据点采样频率从秒级到分钟级不等。一个小区上千户每天产生的原始事件量很容易冲到千万级甚至亿级。这时候用 Python 脚本读文件、写数据库那套做法会直接崩掉——不是代码写错是架构撑不住。「基于 Spark 和 Kafka 的智能家居数据分析系统」这个标题拆开看就是一条标准的数据管道Kafka 负责把散落在各个网关、MQTT Broker、设备云的事件流稳定地接进来充当缓冲和削峰层Spark 负责把这些流式或批量的数据做清洗、聚合、指标计算最后落到存储或看板。它解决的核心问题是「高并发写入 低延迟分析」这对矛盾适合做物联网平台、智慧社区、能耗管理这类场景的工程师也适合想拿一个完整项目练 Spark 实战和 Kafka 消费端调优的人。我见过太多人卡在两个地方一是 Kafka 消费端多线程下消息顺序乱了二是 Spark 内存参数没调任务跑一半 OOM。这篇就按「先跑通、再调优、最后避坑」的顺序把这条管道讲透。2. 数据管道怎么搭Kafka 接入层与 Spark 消费层的职责划分2.1 为什么智能家居场景优先选 Kafka 而不是 RabbitMQ消息队列选型是这套系统的第一个决策点。RabbitMQ、RocketMQ、Kafka 都能做消息中间件但智能家居的数据特征决定了 Kafka 更合适事件量大、写入吞吐要求高、允许一定延迟、消费方可能有多个实时告警、离线分析、冷备归档。Kafka 的核心优势在于分区顺序写磁盘 零拷贝单分区顺序写能到几十 MB/s横向扩分区就能线性提升吞吐。RabbitMQ 在万级 QPS 以下很舒服但队列堆积后性能下降明显RocketMQ 事务消息和延迟消息更强适合电商交易场景。智能家居不需要事务消息需要的是「海量事件不丢、能重放、多消费组独立消费」这正好是 Kafka 的强项。选型上我一般这样定设备事件 topic 按home-events-{region}命名分区数按峰值吞吐除以单分区处理能力估算通常 612 个分区起步。设备状态类数据用 compact topic 保留最新值原始事件用带时间戳的普通 topic保留 7 天。2.2 Kafka 生产端接入从 MQTT 网关到 topic 的桥接设备侧通常走 MQTT网关或边缘服务订阅 MQTT 后转发到 Kafka。下面是一个最小可跑的 Python 生产端模拟网关把设备事件推入 Kafkafrom kafka import KafkaProducer import json, time, random # 关键参数说明 # bootstrap_serversKafka 集群地址生产环境写 3 个 broker # acksall所有 ISR 副本确认后才算成功防丢消息 # linger_ms20攒批 20ms 再发提升吞吐 # compression_typelz4压缩降低网络和磁盘压力 producer KafkaProducer( bootstrap_servers[kafka1:9092, kafka2:9092, kafka3:9092], value_serializerlambda v: json.dumps(v).encode(utf-8), key_serializerlambda k: k.encode(utf-8) if k else None, acksall, linger_ms20, compression_typelz4, retries5 ) def gen_event(device_id): return { device_id: device_id, home_id: device_id.split(-)[0], type: random.choice([temp, humidity, power, motion]), value: round(random.uniform(15, 35), 2), ts: int(time.time() * 1000) } # 用 device_id 做 key保证同一设备的事件进同一分区顺序不乱 for i in range(10000): did fhome{random.randint(1,50)}-dev{random.randint(1,20)} producer.send(home-events, keydid, valuegen_event(did)) producer.flush()这段代码里最容易被忽略的是key。用device_id做 keyKafka 会按 key 哈希到固定分区同一设备的事件天然有序。如果 key 传 None消息轮询进各分区消费端再想按设备聚合就得自己排序代价很大。acksall配合retries是防丢的基本盘代价是延迟略高智能家居场景完全能接受。2.3 Spark 消费端Structured Streaming 读取 Kafka 的最小闭环Spark 侧我优先用 Structured StreamingAPI 统一、支持 exactly-once、和批处理代码几乎一致。下面是从 Kafka 读流、解析 JSON、按设备类型做窗口聚合的最小闭环from pyspark.sql import SparkSession from pyspark.sql.functions import from_json, col, window, avg from pyspark.sql.types import StructType, StringType, DoubleType, LongType spark SparkSession.builder \ .appName(SmartHomeStreaming) \ .config(spark.sql.shuffle.partitions, 12) \ .getOrCreate() schema StructType() \ .add(device_id, StringType()) \ .add(home_id, StringType()) \ .add(type, StringType()) \ .add(value, DoubleType()) \ .add(ts, LongType()) raw spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, kafka1:9092,kafka2:9092,kafka3:9092) \ .option(subscribe, home-events) \ .option(startingOffsets, latest) \ .option(maxOffsetsPerTrigger, 100000) \ .load() parsed raw.select( from_json(col(value).cast(string), schema).alias(d) ).select(d.*).withColumn(event_time, (col(ts) / 1000).cast(timestamp)) agg parsed \ .withWatermark(event_time, 2 minutes) \ .groupBy(window(event_time, 5 minutes), type) \ .agg(avg(value).alias(avg_value)) query agg.writeStream \ .outputMode(update) \ .format(console) \ .option(checkpointLocation, /data/checkpoint/smarthome) \ .trigger(processingTime30 seconds) \ .start() query.awaitTermination()几个参数必须说清楚。maxOffsetsPerTrigger控制每个微批拉多少条防止一次拉太多把内存打爆按集群内存和单条大小估算一般 5 万到 20 万之间。withWatermark处理乱序数据智能家居网络抖动时事件可能晚到2 分钟水位线是经验值。checkpointLocation是 exactly-once 的后悔药丢了它重启就会重复消费。spark.sql.shuffle.partitions默认 200小集群上会开一堆空任务按核数 23 倍设。3. 把原始事件变成指标清洗、聚合与落库的完整链路3.1 数据清洗脏数据在智能家居里长什么样真实设备数据脏得超乎想象。常见的有value 为 null 或负数传感器故障、ts 是 1970 年设备没同步时间、device_id 重复上报、type 字段大小写混用。清洗要在聚合之前做否则平均值会被异常值带偏。from pyspark.sql.functions import when, lower, trim cleaned parsed \ .filter(col(device_id).isNotNull()) \ .filter(col(value).between(-50, 200)) \ .filter(col(ts) 1600000000000) \ .withColumn(type, lower(trim(col(type)))) \ .dropDuplicates([device_id, ts])between(-50, 200)是物理合理区间温度湿度功率都落在这里面超出基本是故障。ts 1600000000000过滤掉 2020 年之前的时间戳能挡掉大部分未同步设备。dropDuplicates按设备和时间戳去重解决网关重发问题。这几步看着简单但少了任何一个后面看板上的曲线都会出现莫名其妙的尖刺。3.2 窗口聚合5 分钟均值、15 分钟峰值怎么算智能家居的分析需求通常分两类实时看板要短窗口15 分钟能耗报表要长窗口小时、天。Structured Streaming 支持滑动窗口和滚动窗口用window函数指定。from pyspark.sql.functions import max, min, count # 5 分钟滚动窗口算均值15 分钟窗口算峰值 windowed cleaned \ .withWatermark(event_time, 3 minutes) \ .groupBy(window(event_time, 5 minutes), home_id, type) \ .agg( avg(value).alias(avg_val), max(value).alias(max_val), min(value).alias(min_val), count(*).alias(cnt) ) # 输出到 Kafka 供下游看板消费 out windowed.selectExpr( to_json(struct(home_id, type, avg_val, max_val, min_val, cnt, window.start as win_start)) as value ) out.writeStream \ .format(kafka) \ .option(kafka.bootstrap.servers, kafka1:9092) \ .option(topic, home-metrics) \ .option(checkpointLocation, /data/checkpoint/metrics) \ .outputMode(update) \ .start()窗口大小和滑动步长的选择有讲究。5 分钟滚动窗口意味着每 5 分钟出一个结果延迟可接受且计算量小。如果要做实时告警用 1 分钟窗口加trigger(processingTime10 seconds)代价是任务数变多。outputMode(update)只输出有更新的窗口比complete省资源但下游要能处理同一窗口的多次更新。3.3 落库选型ClickHouse、HBase 还是直接写 Kafka聚合结果往哪写取决于下游怎么用。实时看板走 Kafka 到前端 WebSocket 最轻历史查询用 ClickHouse写入快、聚合查询快适合按时间范围拉指标设备最新状态用 HBase 或 Redis按 device_id 点查。我一般三路并行Kafka 给实时ClickHouse 给报表Redis 给状态。ClickHouse 写入用 JDBC 或官方 spark-clickhouse-connector注意攒批单批 1 万到 10 万行太小写入放大太大内存压力大。表引擎用ReplacingMergeTree按窗口时间去重避免 Spark 重算导致的重复行。4. 避坑与排查Kafka 消费顺序、Spark 内存和 checkpoint 的五个血泪教训4.1 消费端多线程导致消息顺序错乱现象同一设备的事件在聚合结果里时间倒序均值算出来明显不对。原因为了提升消费速度在消费端开了线程池多个线程并发处理同一分区的消息处理完成顺序和拉取顺序不一致。解决Kafka 的顺序保证只在分区内有效要顺序就一个分区一个消费线程或者用 key 把同一设备路由到同一分区消费端按分区串行处理。Structured Streaming 天然按分区顺序处理别自己再套线程池。4.2 Spark 任务 OOMexecutor 内存和 shuffle 分区没配对现象任务跑十几分钟后 executor 报java.lang.OutOfMemoryError或者 GC 时间超过计算时间。原因spark.sql.shuffle.partitions默认 200小集群上每个分区数据量不均个别分区特别大同时 executor 内存给太小shuffle 数据放不下。解决分区数按总核数 × 23设executor 内存按数据量估一般 48G堆外内存spark.memory.offHeap.enabledtrue配合offHeap.size能缓解。用 Spark UI 的 Stage 页面看 shuffle spill有 spill 就加内存或加分区。4.3 checkpoint 目录丢失导致重复消费现象任务重启后看板数据翻倍同一时间段出现两条记录。原因checkpoint 目录被清理或挂载盘掉了Spark 找不到 offset 就从startingOffsets重新开始。解决checkpoint 放可靠存储HDFS 或云对象存储别放本地盘监控 checkpoint 目录的写入下游存储用幂等写入比如 ClickHouse 的 ReplacingMergeTree 或按窗口时间做主键去重。4.4 Kafka 消息延迟高不是 broker 慢是消费端处理慢现象监控显示 Kafka 堆积量持续上涨消费 lag 越来越大。原因消费端每条消息都同步写数据库单条 RT 几十毫秒吞吐上不去。解决消费端攒批写或者把写库改成异步再不行就加分区加消费者。注意消费者数不能超过分区数多出来的会空转。用kafka-consumer-groups.sh --describe看每个分区的 lag定位是全局慢还是个别分区慢。4.5 时间戳字段类型踩坑毫秒和秒混用现象窗口聚合结果为空或者窗口时间对不上。原因设备上报的 ts 有的是秒级有的是毫秒级/1000之后一个变成 1970 年一个正常。解决接入层统一转成毫秒Spark 里判断位数13 位当毫秒10 位乘 1000。这个坑不报错只是结果静默错误最难查。5. 进阶技巧用 Spark 内存监测和背压把管道跑稳管道跑通只是开始长期稳定运行要靠监测和自适应。Spark 的内存监测我常用两个手段一是 Spark UI 的 Executor 页面看 storage memory 和 execution memory 的占用比例execution 长期打满说明 shuffle 太重二是开spark.eventLog.enabledtrue用 History Server 回看历史任务的内存曲线对比不同参数下的表现。背压方面Structured Streaming 从 Spark 2.3 起支持spark.streaming.backpressure.enabled但 Kafka source 更推荐用maxOffsetsPerTrigger手动限流比自动背压更可控。我一般先按峰值吞吐的 1.5 倍设一个值观察几个批次的处理时间如果每批处理时间远小于 trigger 间隔就调大如果接近或超过就调小或加资源。一个具体技巧把maxOffsetsPerTrigger和trigger间隔联动调。比如 trigger 30 秒单批处理能力 10 万条那maxOffsetsPerTrigger设 10 万保证每批刚好处理完不堆积。这个值要压测得出别拍脑袋。验证管道是否可靠我会做三件事一是故意 kill 掉一个 executor看任务能否自动恢复且不丢不重二是往 Kafka 灌一批带乱序时间戳的数据看水位线是否正确丢弃过期数据三是对比实时聚合结果和离线批处理结果差异在 1% 以内才算过关。最后说个我自己的习惯每次调完参数把配置和对应的监控截图存一份标注日期和场景。Spark 和 Kafka 的参数是玄学同一个值在不同数据分布下表现完全不同没有这份记录下次出问题只能从头试。这套系统值不值得做取决于你的数据量——日事件量低于百万用单机数据库加定时任务更省事上了千万级Kafka 加 Spark 这套组合就是绕不开的基本功。希望帮到你。本文还有配套的精品资源点击获取