基于Spark和Kafka的智能家居数据分析系统源码拆解

简介一份面向智能家居数据开发者的完整实战源码包围绕Spark与Kafka构建端到端数据分析链路适合正在学习实时计算、消息队列或物联网数据处理的读者。资源共16个文件压缩包仅174KB包含Docker编排配置、Spark提交脚本、Kafka/Hadoop环境变量、Mosquitto配置与数据库文件、DDL建表SQL、Arduino主控程序以及Python处理任务脚本覆盖从设备数据采集、消息传输到存储计算和可视化展示的主要环节。已有53人学习下载。使用时可结合Docker快速拉起环境通过SQL与配置文件理解数据表结构和消息桥接方式借助源码对照Spark任务实现实时仪表板与统计分析。整体结构紧凑适合高职、本科毕业设计或个人项目参考也能作为智能家居数据系统二次开发的基础模板。1. 基于Spark和Kafka的智能家居数据分析系统一份能跑通全链路的物联网数据分析源码拆解第一次打开这个zip时我看到的既有main.ino这种单片机固件也有docker-compose.yml和spark-submit.sh就知道这不是一张好看的系统架构图而是一条从DHT温湿度传感器到Web仪表板的完整数据分析链路。标题里的Spark、Kafka、智能家居、数据分析四个词落到这份源码里就是MQTT采集、Kafka缓冲、Spark流处理、PostgreSQL落库、NiFi可视化这条真实可跑的管道。它解决的核心问题只有一个智能家居数据怎么从传感器一路变成能查、能看、能统计的结果。适合手里有ESP8266这类开发板想自建监控系统的人也适合课程设计需要复现一套实时流处理项目的人。2. 整体架构与数据链路MQTT、Kafka、Spark、PostgreSQL 四个环节怎么串起来2.1 设备端采集main.ino 的固件逻辑与传感器库设备端代码放在device_source_code/main/main.ino。从三份依赖库来看——DHT_sensor_library-1.4.4、Adafruit_Unified_Sensor-1.1.12、pubsubclient-master——这是非常典型的Arduino生态组合DHT库负责读取温湿度传感器Adafruit Unified Sensor封装了统一的传感器读取接口PubSubClient负责MQTT通信。以ESP8266为例固件主体长这样#include DHT.h #include PubSubClient.h #include ESP8266WiFi.h #define DHTPIN 4 // GPIO4 接 DHT11/DHT22 数据脚 #define DHTTYPE DHT11 // 传感器型号DHT22 则改成 DHT22 const char* mqtt_server 192.168.1.100; const char* topic home/device1/temperature; WiFiClient espClient; PubSubClient client(espClient); DHT dht(DHTPIN, DHTTYPE); void setup() { Serial.begin(115200); WiFi.begin(your-ssid, your-password); while (WiFi.status() ! WL_CONNECTED) delay(500); client.setServer(mqtt_server, 1883); dht.begin(); } void reconnect() { while (!client.connected()) { if (client.connect(esp8266-dev1)) break; delay(2000); } } void loop() { if (!client.connected()) reconnect(); client.loop(); float h dht.readHumidity(); float t dht.readTemperature(); if (!isnan(h) !isnan(t)) { char payload[64]; snprintf(payload, sizeof(payload), {\device_id\:\dev1\,\temp\:%.2f,\hum\:%.2f}, t, h); client.publish(topic, payload); } delay(5000); // 5 秒上报一次 }这段固件做的事很直接连上WiFi后每5秒读一次DHT传感器把温湿度拼成JSON字符串发布到MQTT broker的home/device1/temperature主题。之所以用JSON而不是裸数值是因为后面Kafka和Spark解析时需要一个自描述的payload以后要加设备状态、电量字段也不用改协议。几个参数值得单独拎出来说。DHTTYPE宏指明传感器型号DHT11和DHT22的时序不一样选错读出来全是NaN。mqtt_server在Docker场景下要填宿主机IP或mosquitto服务名不能填localhost因为开发板和容器不在同一个网络命名空间。topic命名建议带上device_id像home/device1/temperature这样后面Kafka消费者可以用home/#通配符一次订阅所有设备省去逐个主题配置的麻烦。delay(5000)是简单粗暴的上报周期控制实际项目里可以考虑用millis()做非阻塞定时但课程设计和小型监控用delay完全够。这里有一个我踩过的坑DHT库在读取瞬间偶发失败太常见了返回值是NaN直接publish就相当于往整条管道里灌脏数据。代码里isnan(h)和isnan(t)两个判断就是干这个的不要省略。2.2 消息缓冲层Kafka 为什么站在 MQTT 后面MQTT broker用的是mosquitto配置在mosquitto.conf里账号密码放在config/pwfile持久化的消息状态落在mosquitto.db。mosquitto和Kafka之间没有天然连接需要在中间做一次桥接。压缩包里没有单独放桥接程序最常见的做法有两种一是用mosquitto的bridge配置把消息转发到远端二是起一个几行的Python脚本订阅MQTT主题再写入Kafka。第二种更直观也方便在转发过程中做格式修正# mqtt_to_kafka.py import paho.mqtt.client as mqtt from kafka import KafkaProducer # Kafka 生产端value 按 UTF-8 编码存储 producer KafkaProducer( bootstrap_serverskafka:9092, value_serializerlambda v: v.encode(utf-8) ) def on_message(client, userdata, msg): # 把 MQTT 消息原样转发到 Kafka 的 home-sensor-data topic producer.send(home-sensor-data, valuemsg.payload) client mqtt.Client() client.on_message on_message client.connect(mosquitto, 1883) client.subscribe(home/#) # 通配订阅所有设备主题 client.loop_forever()这段脚本里paho.mqtt.client负责连接mosquitto的1883端口KafkaProducer负责往Kafka broker写数据。on_message回调里每收到一条MQTT消息就同步发送到Kafkavalue_serializer把字节转成UTF-8字符串。subscribe(home/#)用通配符订阅全部设备比逐个主题订阅省事得多。有人会问设备端直接发Kafka不行吗现实里不行。智能家居设备通常跑在WiFi环境里MQTT的发布订阅模型对弱网和断线重连更友好协议开销也更小Kafka是为大数据吞吐设计的客户端比较重不适合单片机直接跑。所以MQTT接设备、Kafka接平台的组合是这类场景里经过验证的分工中间这一层桥接虽然多一跳但换来的是设备端极简和平台端稳定。2.3 存储与计算层HDFS、Spark、PostgreSQL 各自守哪一段HDFS、Spark、PostgreSQL三个组件听名字容易混实际上各守一段。HDFS是分布式文件系统负责存原始数据和Spark检查点保证流处理重启不丢状态Spark从Kafka消费数据做清洗、开窗、聚合这是整个系统里计算密度最高的地方PostgreSQL只存聚合结果供Web仪表板查询。简单说HDFS是数据湖Spark是计算引擎PostgreSQL是结果库。PostgreSQL的表结构在ddl.sql里定义典型的指标统计宽表长这样CREATE TABLE device_metrics ( id SERIAL PRIMARY KEY, device_id VARCHAR(32) NOT NULL, window_start TIMESTAMP NOT NULL, window_end TIMESTAMP NOT NULL, avg_temp DOUBLE PRECISION, max_temp DOUBLE PRECISION, min_temp DOUBLE PRECISION, avg_hum DOUBLE PRECISION, created_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX idx_metrics_device_window ON device_metrics(device_id, window_start);窗口起止时间、设备ID、各指标聚合值是这张表的灵魂。复合索引idx_metrics_device_window专门加速“查某设备某时间段”的请求Web面板首页的曲线图基本都是走这个索引。注意这里没有保留原始数据原始数据在HDFS里PostgreSQL表膨胀了可以直接清重跑Spark作业就能回填这就是流批一体带来的操作空间。为什么不直接查明细因为传感器5秒一条一天下来一台设备就有17000多条记录Web面板需要的是秒级响应的聚合值而不是几十万行的明细。这个取舍是数据分析项目里最常见也最需要提前想明白的设计决策。3. Docker 环境搭建与启停docker-compose.yml 的编排顺序与调参3.1 服务编排清单与端口规划从压缩包里的mosquitto.conf、hadoop.env、docker-compose.yml、nifi_registry目录来看这份编排至少涉及六个服务。我第一次部署时没做端口规划结果8080端口被本地一个程序占了排查了半天才知道是端口冲突。先看整体清单服务容器内端口宿主机端口作用mosquitto18831883MQTT broker接收设备上报hadoop-namenode98709870HDFS NameNode管理元数据hadoop-datanode98649864HDFS DataNode存原始数据kafka90929092消息队列缓冲传感器数据spark-master7077 / 80807077 / 8080Spark主节点作业提交入口spark-worker80818081Spark工作节点执行计算postgres54325432结果数据库nifi-registry1808018080可视化流程注册中心表格里宿主机端口是按默认情况写的实际以你本机的docker-compose.yml为准。端口规划有三条原则宿主机端口不要和本机已监听端口冲突Kafka的9092对外暴露时容器内的advertised.listeners要写成宿主机IP或服务名否则Spark容器里连不上nifi-registry的18080和Spark的8080别搞混一个是Web UI一个是集群管理端口。3.2 启动顺序与健康检查Kafka 没就绪不能提作业这份系统的依赖关系是Kafka要等ZookeeperSpark要等Kafka和PostgreSQLmosquitto和nifi-registry最后起。我一般不用一条docker compose up -d梭哈而是分两段启动每段之间做端口探测# 第一步先把基础设施拉起来 docker compose up -d zookeeper kafka postgres # 轮询等 Kafka 和 PostgreSQL 就绪 until docker exec $(docker compose ps -q kafka) \ /opt/kafka/bin/kafka-broker-api-versions.sh \ --bootstrap-server localhost:9092 /dev/null 21; do sleep 3 done sleep 5 # 第二步启动计算和可视化服务 docker compose up -d spark-master spark-worker mosquitto nifi-registry # 确认所有容器都是 healthy docker compose ps这段脚本里docker compose ps -q kafka拿到Kafka容器IDkafka-broker-api-versions.sh是Kafka自带的管理脚本能连上broker就说明Kafka真正就绪而不是端口通就算。这里有个容易翻车的点很多人用nc -z localhost 9092去测但Kafka的端口在listener初始化完成前就可能已经监听了测出来是通的Spark一提交作业却报连接超时。所以宁可多等几秒也别省这一步。提示容器环境下所有跨服务连接的主机名都写成docker-compose里的服务名比如kafka、postgres、mosquitto而不是localhost。3.3 启动阶段最容易翻车的三个点启动阶段的问题往往和环境有关和代码关系不大但一旦卡住就很劝退。我整理三个频率最高的第一个是mosquitto容器起不来。现象是docker compose logs里有Address already in use或权限错误。原因一般是宿主机1883端口被本机同名服务占用或者mosquitto/data目录的属主不对容器内mosquitto用户写不进去。解决方法是先执行netstat -lnp | grep 1883看占用进程改端口映射挂载目录先chown -R 1883:1883或用chmod 777解决读写权限。第二个是Spark容器内存不足。现象是spark-master健康检查一直不过日志里出现Native memory allocation failed。原因是默认给worker分配的2G内存超过Docker Desktop或云主机的可用额度。解决方法是把docker-compose.yml里的SPARK_WORKER_MEMORY改成1g同时给Docker引擎多分一点内存。第三个是容器间互相访问不通。现象是Spark作业里写jdbc:postgresql://localhost:5432连不上但宿主机上psql连同一个库是好的。原因是localhost在容器内指向容器自己跨容器访问必须用服务名。解决方法是把JDBC URL改成jdbc:postgresql://postgres:5432/smarthomeKafka地址同步改成kafka:9092。这个问题在刚上手Docker时几乎必踩记住“容器内没有localhost”这句话能省很多时间。4. 数据处理作业解析process_job.py 与 spark-submit.sh 的正确打开方式4.1 从 Kafka 读流结构化流的连接参数与 offset 策略process_job_lib/process_job.py是系统的计算核心。第一步是从Kafka读流Spark结构化流用readStream方式声明数据源from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, StringType, DoubleType from pyspark.sql.functions import from_json, col spark SparkSession.builder \ .appName(SmartHomeStreaming) \ .config(spark.sql.shuffle.partitions, 4) \ .getOrCreate() # 设备和 payload 里的 JSON 结构保持一致 payload_schema StructType([ StructField(device_id, StringType()), StructField(temp, DoubleType()), StructField(hum, DoubleType()) ]) raw spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, kafka:9092) \ .option(subscribe, home-sensor-data) \ .option(startingOffsets, latest) \ .option(failOnDataLoss, false) \ .load()读出来的DataFrame里value列是Kafka消息的原始字节另外还有topic、partition、offset等元数据列。要从JSON payload里取字段得先转字符串再from_jsonparsed raw.selectExpr(CAST(value AS STRING) AS json) \ .select(from_json(col(json), payload_schema).alias(data)) \ .select(data.device_id, data.temp, data.hum)这段里面bootstrap.servers写kafka:9092而不是localhost:9092原因在前面已经说过容器内必须用服务名。startingOffsets有latest和earliest两个值latest只消费作业启动后的新数据适合在线监控earliest会从头消费topic里的历史数据适合补数场景。failOnDataLoss一定要设成false否则Kafka topic因清理策略被删除后作业重启会直接抛出OffsetOutOfRangeException这是流作业最常见的崩溃原因之一。4.2 清洗、窗口与聚合事件时间怎么处理接下来是数据清洗。MQTT转Kafka那一步如果固件里的isnan没挡住或者传感器在异常环境里读出负温度脏数据就进到了流里。清洗要做三件事过滤空值、过滤合理范围、补上事件时间。from pyspark.sql.functions import current_timestamp, window, avg, max, min cleaned parsed.filter(col(temp).isNotNull()) \ .filter((col(temp) -40) (col(temp) 80)) \ .filter((col(hum) 0) (col(hum) 100)) \ .withColumn(event_time, current_timestamp()) aggregated cleaned.withWatermark(event_time, 1 minutes) \ .groupBy(window(col(event_time), 5 minutes), col(device_id)) \ .agg( avg(temp).alias(avg_temp), max(temp).alias(max_temp), min(temp).alias(min_temp), avg(hum).alias(avg_hum) )这里有几个细节要说明。temp过滤范围-40到80摄氏度这是DHT系列传感器的量程定义低于-40或高于80本质上就是读取错误不是真实环境温度。hum的0到100同理超出这个范围的数据没有物理意义。withColumn(event_time, current_timestamp())是给每条记录打上处理时间因为固件发的JSON里没有设备时间戳这是物联网场景下的常见妥协。如果以后你在固件payload里加了time字段就要把这里换成to_timestamp(col(time), yyyy-MM-dd HH:mm:ss)否则窗口会全部按当前时间计算离线补数时就对不上了。withWatermark(event_time, 1 minutes)有两个作用明确允许迟到1分钟内的数据进入窗口计算同时让Spark定期清理过期状态防止流作业跑几天后内存炸掉。这个参数的大小要和业务容忍度匹配设太大状态膨胀设太小晚到的数据被丢弃。窗口函数window(..., 5 minutes)做的是滚动窗口每隔5分钟汇总一次想改成10分钟就改这一个参数Web面板的柱状图粒度也跟着变。groupBy里device_id是维度窗口是时间维度这两个字段组成聚合的粒度和2.3里device_metrics表的主键约定先对好。4.3 写回 PostgreSQLforeachBatch 与 JDBC 参数聚合结果不能直接落库因为DataFrame里window是一个结构体列而PostgreSQL表里要求window_start和window_end两个独立时间字段。foreachBatch模式在每批微批数据上执行一次外部写操作比逐行写入效率高很多def write_to_postgres(batch_df, batch_id): batch_df.select( col(device_id), col(window.start).alias(window_start), col(window.end).alias(window_end), col(avg_temp), col(max_temp), col(min_temp), col(avg_hum) ).write \ .mode(append) \ .format(jdbc) \ .option(url, jdbc:postgresql://postgres:5432/smarthome) \ .option(dbtable, device_metrics) \ .option(user, postgres) \ .option(password, postgres) \ .option(Driver, org.postgresql.Driver) \ .option(batchsize, 1000) \ .save() query aggregated.writeStream \ .foreachBatch(write_to_postgres) \ .outputMode(update) \ .trigger(processingTime30 seconds) \ .start() query.awaitTermination()batchsize1000表示JDBC批量大小调太高会给PostgreSQL造成压力写放大明显太低则吞吐上不去一秒只能写几百行。trigger(processingTime30 seconds)表示每30秒触发一次微批和传感器5秒的采集周期相比一个窗口内会有多批数据到达但落库的是窗口合并后的结果不会重复。outputMode(update)配合foreachBatch是推荐组合append模式在这种带聚合的流查询上部分版本会报不支持遇到就换update。代码里.option(Driver, org.postgresql.Driver)这行容易被忽略但如果不写某些Spark发行版会因为加载不到驱动而报No suitable driver显式声明能让问题早点暴露。4.4 提交作业spark-submit.sh 的依赖与资源参数process_job.py要跑在Spark集群上提交脚本长这样版本号按你本地Spark版本改#!/usr/bin/env bash SPARK_MASTERspark://localhost:7077 SCALA_BINARY2.12 SPARK_VERSION3.3.0 PG_DRIVERorg.postgresql:postgresql:42.5.1 spark-submit \ --master ${SPARK_MASTER} \ --deploy-mode client \ --driver-memory 512m \ --executor-memory 1g \ --total-executor-cores 2 \ --packages org.apache.spark:spark-sql-kafka-0-10_${SCALA_BINARY}:${SPARK_VERSION},${PG_DRIVER} \ process_job_lib/process_job.py这个脚本里最容易漏的是--packages。spark-sql-kafka-0-10是Spark读写Kafka的官方连接器postgresql是JDBC驱动不通过--packages预置的话作业一运行就报ClassNotFoundException。注意artifactId里的_2.12是Scala版本号必须和Spark发行版一致装了Scala 2.13的Spark却写2.12运行时会报类冲突或NoSuchMethodError。executor-memory和total-executor-cores根据机器配置调小内存机器给512m和1核也能跑只是吞吐差一些。deploy-mode client方便在终端直接看日志排错阶段强烈建议先用client模式跑通再考虑cluster模式。这里还有一个spark集群搭建时很容易忽视的点--packages下载依赖需要网络访问Maven中央仓库如果内网环境没有代理要提前把jar包放到SPARK_HOME/jars目录里否则提交阶段会卡很久然后超时。5. 避坑与常见问题排查五条高频坑的根因与解法下面五条是我在复现和调整这类智能家居流处理项目时真实踩过的按出现频率排序每一条都照着“现象、原因、解决”来写。5.1 传感器数据偶发 NaN下游图表出现断点现象Web仪表板的温度曲线每隔一段时间出现一个缺口PostgreSQL里偶发null值Kafka里能看到NaN对应的无效记录。原因DHT传感器在湿度较大或电流不稳时读取会失败库函数返回NaN固件里没有isnan判断就直接publish。这是硬件层面的随机故障跟代码逻辑无关。解决固件里加上isnan过滤这是第一道防线Spark清洗层再过滤一次范围这是第二道防线。两层都做脏数据基本进不了结果表。我在main.ino和process_job.py里都保留了这两层逻辑不建议省掉任意一层。5.2 Kafka 消费 Lag 只涨不跌面板数据越来越旧现象在Kafka可视化工具里看消费者组的Lag持续上涨仪表板上的数据落后真实时间十几分钟而且没有恢复的迹象。原因要么Spark作业的处理能力跟不上生产速度要么作业重启后startingOffsets设了earliest导致从头消费历史数据。生产速率模拟出来很容易处理链路里的shuffle和JDBC写入才是瓶颈。解决先用kafka-consumer-groups.sh --describe --group smart_home_group看各分区Lag分布确认是不是某个分区积压特别严重然后确认startingOffsets是latest最后调大executor-memory、增加total-executor-cores同时把spark.sql.shuffle.partitions从默认200调小到和topic分区数匹配比如4减少shuffle开销。这里典型的血泪教训是只调大内存不调分区数结果shuffle那一层还是堵。5.3 Spark 写 PostgreSQL 报 Connection refused但 psql 能连现象Spark作业在foreachBatch里报Connection to postgres:5432 refused但宿主机上psql连接同一个库是好的数据用DBeaver也能查到。原因容器内的localhost指向容器自己JDBC URL必须用docker-compose服务名替换。宿主机能连是因为端口映射到了宿主机容器内访问localhost走的是容器自己的网络栈。解决把jdbc:postgresql://localhost:5432改成jdbc:postgresql://postgres:5432Kafka的bootstrap.servers同理改成kafka:9092。这个坑在从非容器环境迁移到Docker环境时最容易遇到我见过有人排查了一个下午最后发现只是主机名没换。5.4 NiFi Registry Web UI 打不开数据库文件锁报错现象docker compose up后nifi-registry容器反复重启日志里报H2 database file is locked或者页面打开后一直转圈。原因nifi-registry容器没有正常停止H2数据库的.lock文件残留或者宿主机上有另一个Java进程占用了nifi-registry数据库目录下的nifi-registry-primary.mv.db文件。这个文件是整个NiFi流程配置的黑匣子出问题时的表现非常隐蔽。解决先docker compose stop nifi-registry删除挂载目录下nifi_registry/database里的*.lock.db文件再重新启动。提交NiFi流程之前先手动备份那个mv.db文件这是我吃过亏之后保留的习惯遇到数据库文件损坏时备份就是唯一的后悔药。5.5 MQTT 密码认证失败设备连不上 mosquitto现象设备串口打印Connection refused或Not authorizedmosquitto容器日志里出现Bad username or password。原因config/pwfile是用mosquitto_passwd生成的但mosquitto.conf里要么没有写password_file指令要么allow_anonymous没有设成false导致broker仍然期望匿名连接和客户端的用户名密码逻辑对不上。解决检查mosquitto.conf里是否有password_file /path/to/pwfile和allow_anonymous falsepwfile重新生成时用mosquitto_passwd -c pwfile username命令。注意pwfile文件的权限要设成600mosquitto对配置文件权限很敏感权限太开放反而会拒绝启动。6. 验证链路与进阶玩法对照 schema.jpg 做血缘检查再给面板加一条告警系统跑通后别急着去看界面样式先把数据血缘对一遍。压缩包里的artifacts/schema.jpg是整条链路的关系图我一般按这个顺序核对先用Kafka自带命令确认topic里有数据# 确认 MQTT 桥接是否把数据写进了 Kafka kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic home-sensor-data \ --from-beginning \ --max-messages 5能看到JSON输出说明MQTT到Kafka这一段是通的。接着查PostgreSQL的device_metrics表确认窗口聚合结果在持续写入SELECT device_id, window_start, window_end, avg_temp, avg_hum FROM device_metrics ORDER BY window_start DESC LIMIT 5;如果Kafka有数据而表里没有问题基本出在Spark作业如果表里有数据但面板不显示问题出在Web后端的查询SQL。对照schema.jpg可以把问题快速定位到具体环节这个习惯比任何调试工具都管用。在此基础上一个很实用的进阶玩法是加告警规则。比如设备温度连续两个窗口超过40度就报警不需要改Spark作业直接在Web后端加一条SQL检测SELECT device_id, MAX(max_temp) AS peak_temp FROM device_metrics WHERE window_start NOW() - INTERVAL 10 minutes GROUP BY device_id HAVING MAX(max_temp) 40;这条SQL每5分钟跑一次有结果就触发通知相当于用已有的聚合表搭了一个最简阈值告警不用为了告警单独拉一套规则引擎。想更精细可以做两级阈值超过35度记录到日志超过40度才发通知避免半夜被误报吵醒。从那以后我每次部署这类流处理项目都强制自己按schema.jpg把整条链路的血缘走一遍Kafka有数据、表在写、面板在刷三步确认完才关终端。这套动作救过我很多次希望帮到你。本文还有配套的精品资源点击获取