简介：这是一套面向物联网与大数据方向学习者的智能家居数据分析系统源码，适合具备一定Spark、Kafka基础、希望动手实践流式数据处理的中高级开发者。项目以MQTT协议采集智能家居设备传感器数据，经Kafka消息队列实现实时传输，再由Spark完成处理分析并写入PostgreSQL，同时提供Web实时仪表板与NiFi数据可视化界面，覆盖从数据采集、存储到分析展示的完整链路。资源包共16个文件，包含zip依赖库、db数据库文件、yml容器编排配置、sql建表脚本、ino设备端程序、py处理脚本及sh启动脚本等，压缩包约174KB，结构紧凑便于快速部署。目前已有53人学习下载。读者可借此掌握Docker环境搭建、Kafka与Spark协同、HDFS存储及仪表板展示的完整实现思路，是理解智能家居实时分析架构的实用参考。
1. 智能家居数据管道：为什么 Spark 加 Kafka 是绕不开的组合
家里几十个传感器，门窗磁、温湿度、人体红外、智能插座，每秒都在吐数据。单机脚本跑得动一百台设备，跑不动一万台；批处理能算昨天的用电报表，算不了“现在客厅有人且温度高于 28 度就联动开空调”。这就是智能家居数据分析最真实的痛点：数据源持续不断、事件有先后顺序、既要实时告警又要离线复盘。Kafka 负责把散落在各处的设备事件可靠地收上来，Spark 负责把流式和批量两条线用同一套代码逻辑算清楚。这套组合不是赶时髦，而是因为设备消息天然是追加日志，Kafka 的分区模型刚好匹配设备 ID 的并行度，Spark Structured Streaming 又能让流处理和批处理共享同一份 DataFrame API。适合谁？适合手里已经有智能家居设备数据、想从“能存能看”走到“能算能控”的工程师，也适合想拿一个完整项目练手 Spark 和 Kafka 协同的数据开发。下面按“先跑通最小链路，再谈参数和坑”的顺序拆开讲。
2. 从设备消息到 Kafka Topic：数据接入层怎么设计
2.1 智能家居事件模型与 Topic 划分
智能家居的数据不像电商订单那样规整。一个温湿度传感器上报的是{"deviceId":"th-001","type":"temperature","value":26.5,"ts":1710000000000}，一个人体红外上报的是{"deviceId":"pir-007","type":"motion","value":1,"ts":1710000000123}。如果所有设备塞进一个 Topic，下游解析时字段对不齐，分区也没法按设备隔离。常见做法是按数据大类拆 Topic：smarthome.sensor.telemetry放周期性遥测，smarthome.device.event放开关、告警、联动触发这类离散事件。分区数怎么定？一个分区只能被一个消费者线程读，分区太少限制并行度，太多增加协调开销。我一般按“峰值每秒消息数 ÷ 单分区每秒可处理消息数”估算，再留一倍余量。比如峰值 5000 条/秒，单分区实测能扛 2000 条/秒，那 3 个分区够用，但为了后续加消费者，设 6 个更稳妥。分区键用deviceId，这样同一设备的消息严格有序，做状态判断时不会因为乱序误判。
2.2 用 Python 模拟设备上报并写入 Kafka
没有真实设备时，先用脚本造数据把链路跑通。下面这段代码模拟 10 个设备，每 0.5 秒发一条温湿度或人体感应消息。
from kafka import KafkaProducer import json, time, random # 连接本地 Kafka，bootstrap_servers 按实际地址改 producer = KafkaProducer( bootstrap_servers=['localhost:9092'], value_serializer=lambda v: json.dumps(v).encode('utf-8'), # 按 deviceId 做分区，保证同一设备消息有序 key_serializer=lambda k: k.encode('utf-8') ) devices = [f'th-{i:03d}' for i in range(5)] + [f'pir-{i:03d}' for i in range(5)] while True: for dev in devices: if dev.startswith('th'): msg = { 'deviceId': dev, 'type': 'temperature', 'value': round(random.uniform(18, 32), 1), 'ts': int(time.time() * 1000) } topic = 'smarthome.sensor.telemetry' else: msg = { 'deviceId': dev, 'type': 'motion', 'value': random.choice([0, 1]), 'ts': int(time.time() * 1000) } topic = 'smarthome.device.event' # key 传 deviceId，Kafka 按 key hash 落到固定分区 producer.send(topic, key=dev, value=msg) producer.flush() time.sleep(0.5)逻辑说明：value_serializer把字典转成 JSON 字节流，key_serializer把设备 ID 转成字节流。Kafka 对相同 key 的消息会分配到同一分区，这是后续做设备级状态计算的前提。producer.flush()在循环里调用是为了确保消息真正发出，生产环境可以靠linger_ms攒批，但调试阶段直接 flush 更直观。参数上，bootstrap_servers如果连的是集群，写两三个 broker 地址用逗号隔开，不要只写一个。acks默认是 1，对智能家居遥测数据够用；如果是门锁开关这类事件，建议设成all，避免 broker 落盘前挂掉丢消息。
2.3 验证消息是否落进正确分区
写完数据别急着写 Spark，先用命令行确认 Topic 和分区分布。
# 列出所有 topic kafka-topics.sh --bootstrap-server localhost:9092 --list # 查看 telemetry topic 的分区详情 kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic smarthome.sensor.telemetry # 从开头消费 10 条看看内容 kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic smarthome.sensor.telemetry --from-beginning --max-messages 10--describe输出里会显示每个分区的 Leader 和 Replica 分布，如果某个分区 Leader 是 -1，说明副本没同步上，先查 broker 日志。--from-beginning只对还没被消费组提交偏移量的情况有效，如果之前有消费者提交过，得加--group指定新组名才能从头读。这一步看着简单，但很多“Spark 读不到数据”的问题，根因是 Topic 名拼错或者消息压根没写进去。
3. Spark Structured Streaming 消费 Kafka：最小可跑通的流处理
3.1 流式读取 Kafka 的 DataFrame 写法
Spark 3.x 之后，Structured Streaming 读 Kafka 就是spark.readStream.format("kafka")一行的事，但参数配不对就会卡在启动阶段。下面是最小可跑通的 PySpark 代码。
from pyspark.sql import SparkSession from pyspark.sql.functions import from_json, col from pyspark.sql.types import StructType, StringType, DoubleType, LongType spark = SparkSession.builder \ .appName("SmartHomeStreaming") \ .config("spark.sql.shuffle.partitions", "6") \ .getOrCreate() # 定义 JSON 消息的 schema，字段要和生产者发的对齐 schema = StructType() \ .add("deviceId", StringType()) \ .add("type", StringType()) \ .add("value", DoubleType()) \ .add("ts", LongType()) raw_df = spark.readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "localhost:9092") \ .option("subscribe", "smarthome.sensor.telemetry") \ .option("startingOffsets", "latest") \ .option("failOnDataLoss", "false") \ .load() # Kafka 的 value 是二进制，先转字符串再解析 JSON parsed_df = raw_df.select( from_json(col("value").cast("string"), schema).alias("data") ).select("data.*") # 简单过滤：只看温度高于 28 度的记录 hot_df = parsed_df.filter(col("type") == "temperature").filter(col("value") > 28) query = hot_df.writeStream \ .outputMode("append") \ .format("console") \ .option("truncate", "false") \ .trigger(processingTime="10 seconds") \ .start() query.awaitTermination()逻辑说明：startingOffsets设成latest表示只读启动之后新到的消息，调试时如果想让历史数据也进来，改成earliest。failOnDataLoss设false是为了在 Kafka 清理了旧 offset 时不至于让整个流挂掉，生产环境要配合监控告警。from_json的 schema 必须和生产者发的 JSON 字段类型严格一致，value在生产者那边是浮点数，这里用DoubleType，如果写成StringType会解析出 null。trigger设 10 秒是为了在控制台看到攒批效果，实际部署可以改成processingTime="1 minute"降低小文件压力。
3.2 检查点目录与输出模式的选择
Structured Streaming 靠 checkpoint 记录消费进度和状态，不配 checkpoint 目录，流任务重启后会从头再来或者直接报错。
query = hot_df.writeStream \ .outputMode("append") \ .format("parquet") \ .option("path", "/data/smarthome/hot_temperature") \ .option("checkpointLocation", "/data/smarthome/checkpoint/hot_temperature") \ .trigger(processingTime="1 minute") \ .start()checkpointLocation必须是一个持久化存储路径，本地调试用文件系统可以，集群上要换成 HDFS 或对象存储。outputMode三种选择：append只输出新行，适合落 parquet；update输出被更新的行，适合做聚合后写数据库；complete每次输出全量结果，只适合结果集很小的聚合。智能家居场景里，原始遥测落盘用append，房间平均温度这种聚合用update写 Redis 或 MySQL。注意 checkpoint 目录一旦启用，不要手动删里面的文件，否则流任务恢复时会报 offset 不连续。
3.3 用 Spark SQL 做窗口聚合：每 5 分钟房间平均温度
流处理不只是过滤，还要做时间窗口聚合。下面按设备分组，算 5 分钟滚动窗口的平均温度。
from pyspark.sql.functions import window, avg windowed_df = parsed_df \ .filter(col("type") == "temperature") \ .groupBy(window(col("ts").cast("timestamp"), "5 minutes"), col("deviceId")) \ .agg(avg("value").alias("avg_temp")) query = windowed_df.writeStream \ .outputMode("update") \ .format("console") \ .option("truncate", "false") \ .option("checkpointLocation", "/data/smarthome/checkpoint/avg_temp") \ .trigger(processingTime="1 minute") \ .start()window函数的第一个参数是时间列，必须是 timestamp 类型，Kafka 消息里的ts是毫秒时间戳，用cast("timestamp")转换。窗口长度 5 分钟，没有设滑动步长，默认就是滚动窗口。outputMode("update")让每个微批只输出有变化的设备聚合结果，避免重复写全量。这里有个容易翻车的点：如果设备上报时间戳用的是设备本地时间且没同步 NTP，窗口会错乱，建议在接入层统一用服务端接收时间做事件时间，或者至少校验设备时间偏差。
4. 离线批处理与流批一体：同一套逻辑跑两条线
4.1 用 Spark SQL 做历史数据复盘
流处理解决实时告警，离线批处理解决报表和模型训练。智能家居的离线分析常见需求是：按天统计每个房间的用电量、找出温度异常时段、分析人体感应与灯光开启的相关性。这些用 Spark SQL 直接查 parquet 文件就行。
# 读昨天落盘的 parquet 数据 batch_df = spark.read.parquet("/data/smarthome/telemetry/dt=2024-03-01") batch_df.createOrReplaceTempView("telemetry") # 按设备统计当天温度最大值、最小值和平均值 spark.sql(""" SELECT deviceId, MAX(value) AS max_temp, MIN(value) AS min_temp, ROUND(AVG(value), 2) AS avg_temp, COUNT(*) AS sample_count FROM telemetry WHERE type = 'temperature' GROUP BY deviceId ORDER BY avg_temp DESC """).show()逻辑说明：parquet 按天分区存储时，路径里带dt=2024-03-01这种分区目录，Spark 读取时会自动做分区裁剪，只扫对应目录。createOrReplaceTempView注册临时视图后就能用纯 SQL 表达，适合数据分析师直接上手。sample_count用来判断数据完整性，如果某个设备当天只有几条记录，说明设备离线或网络有问题，这种异常要在报表里标出来，不能直接拿平均值下结论。
4.2 流批一体：用同一份 schema 和转换逻辑
流处理和批处理最大的重复劳动是 schema 定义和字段清洗。把这两部分抽成独立模块，流和批都引用同一份代码。
# schema_def.py from pyspark.sql.types import StructType, StringType, DoubleType, LongType TELEMETRY_SCHEMA = StructType() \ .add("deviceId", StringType()) \ .add("type", StringType()) \ .add("value", DoubleType()) \ .add("ts", LongType()) def clean_telemetry(df): """统一清洗逻辑：过滤空设备 ID，转换时间戳""" from pyspark.sql.functions import col, to_timestamp return df.filter(col("deviceId").isNotNull()) \ .withColumn("event_time", to_timestamp(col("ts") / 1000))流处理里from_json(col("value").cast("string"), TELEMETRY_SCHEMA)，批处理里spark.read.schema(TELEMETRY_SCHEMA).json(...)，清洗函数两边都调clean_telemetry。这样改字段类型或加过滤条件时只改一处，不会出现流和批算出来的结果对不上。我见过一个项目，流处理里把温度单位从摄氏度改成了华氏度，批处理脚本没同步改，导致实时告警和离线报表差了 32 度，排查了一下午才发现是两套代码。血泪经验就是：能抽公共模块就别复制粘贴。
4.3 把聚合结果写回 Kafka 供下游消费
算完的结果不一定只落库，也可以写回 Kafka 让告警服务、App 推送服务去消费。
result_df = windowed_df.selectExpr( "CAST(deviceId AS STRING) AS key", "to_json(struct(*)) AS value" ) query = result_df.writeStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "localhost:9092") \ .option("topic", "smarthome.agg.room_temp") \ .option("checkpointLocation", "/data/smarthome/checkpoint/agg_to_kafka") \ .outputMode("update") \ .start()写回 Kafka 时，DataFrame 必须包含key和value两列，且都是字符串或二进制。to_json(struct(*))把整行转成 JSON 字符串。outputMode用update是因为聚合结果会变，下游按 key 做幂等更新。注意写回 Kafka 的 Topic 分区数最好和上游一致，否则可能出现下游消费并行度不匹配。另外，如果下游是告警服务，消息里要带窗口起止时间，不然收到“平均温度 30 度”却不知道是哪 5 分钟的数据。
5. 避坑与排查：智能家居流处理里最容易翻车的 4 个点
5.1 现象：Spark 流任务启动后一直没输出，控制台空白
原因：startingOffsets设成了latest，而生产者还没开始发数据，或者 Topic 名写错导致订阅了一个空 Topic。另一个常见原因是trigger设了较长的 processingTime，比如 5 分钟，启动后要等第一个微批结束才输出。解决：调试阶段把startingOffsets改成earliest，trigger改成processingTime="10 seconds"，同时用kafka-console-consumer.sh确认 Topic 里确实有消息。如果 Topic 有消息但 Spark 读不到，检查subscribe参数有没有拼写错误，Kafka 的 Topic 名是大小写敏感的。
5.2 现象：JSON 解析后字段全是 null
原因：from_json的 schema 和实际 JSON 字段类型不匹配。比如生产者发的value是整数26，schema 里定义成DoubleType通常能兼容，但如果生产者发的是字符串"26.5"，schema 里定义成DoubleType就会解析失败返回 null。解决：先用kafka-console-consumer.sh看原始消息长什么样，再对照 schema 逐字段核对。一个省事的办法是先用StringType把所有字段读进来，再用cast转换，这样解析失败时能看到原始字符串，方便定位。
5.3 现象：流任务跑一段时间后报 OffsetOutOfRange
原因：Kafka 的log.retention.hours默认是 168 小时，如果流任务停了超过这个时间，之前提交的 offset 对应的消息已经被清理，重启时就会报 offset 越界。解决：把failOnDataLoss设成false让任务跳过丢失的数据继续跑，同时检查 checkpoint 目录是否被误删。长期方案是调大 retention 时间，或者让流任务保持运行，不要长时间停机。如果业务允许丢数据，startingOffsets设latest重新开始也行，但要评估丢失窗口对业务的影响。
5.4 现象：窗口聚合结果延迟很高，告警不及时
原因：trigger的 processingTime 设得太长，比如 10 分钟，加上窗口长度 5 分钟，最坏情况下要等 15 分钟才输出。另一个原因是spark.sql.shuffle.partitions设得太大，小数据量下反而增加调度开销。解决：实时告警场景把trigger降到 10 到 30 秒，窗口长度按业务容忍度调整。shuffle.partitions在本地调试时设成 6 到 12 就够，集群上按数据量调，一般设成 CPU 核数的 2 到 3 倍。如果还是慢，看 Spark UI 里哪个 stage 耗时最长，通常是 Kafka 读取或 JSON 解析阶段，可以增加maxOffsetsPerTrigger限制每批拉取量，避免单批数据过大。
6. 进阶技巧：用水位线处理迟到数据和验证结果一致性
智能家居设备网络不稳定，消息迟到是常态。Structured Streaming 的水位线机制可以容忍一定程度的迟到，同时控制状态大小。下面在窗口聚合基础上加 10 分钟水位线。
from pyspark.sql.functions import window, avg, col windowed_df = parsed_df \ .withWatermark("event_time", "10 minutes") \ .filter(col("type") == "temperature") \ .groupBy(window(col("event_time"), "5 minutes"), col("deviceId")) \ .agg(avg("value").alias("avg_temp"))withWatermark必须用在事件时间列上，且要在聚合之前调用。10 分钟水位线意味着 Spark 会等 10 分钟，之后到达的、事件时间早于水位线的数据会被丢弃。水位线设太长，状态存储压力大；设太短，迟到数据丢得多。我一般先按设备网络质量估算最大迟到时间，再留一倍余量。比如 4G 设备最坏迟到 3 分钟，水位线设 6 到 10 分钟。
验证结果一致性有个笨但有效的办法：把同一时间段的数据分别用流处理和批处理跑一遍，对比聚合结果。流处理输出到 parquet 后，用批处理读同一份原始数据算一遍，两边按deviceId和窗口起止时间 join，看avg_temp差值是否在浮点误差范围内。如果差得多，先查水位线是不是把迟到数据丢了，再查流处理的 checkpoint 是否从正确 offset 开始。这个对账步骤在项目上线前跑一次，能提前发现大部分逻辑错误。
最后说个习惯：我每次改完流处理代码，不会直接上生产，而是先用earliest从头消费一小段历史数据，确认输出符合预期后再切latest部署。这个习惯帮我省掉了至少三次半夜爬起来回滚的麻烦。希望帮到你。
本文还有配套的精品资源，点击获取
