尧图网页设计 
简介这是一套面向物联网与大数据方向学习者、开发者及课程设计者的智能家居数据分析系统源码围绕设备数据采集、流式传输、分布式存储与实时分析展开帮助读者理解从传感器到可视化仪表板的完整链路。资源包共16个文件包含zip依赖库、db数据库文件、yml容器编排配置、sql建表脚本、conf与env环境配置、ino设备端程序、py数据处理脚本及sh提交脚本等压缩包约174KB结构紧凑、模块划分清晰。项目以MQTT协议采集智能家居设备数据经Kafka消息队列保障实时性与可靠性由HDFS完成大规模存储再通过Spark进行高效处理分析并将结果写入PostgreSQL最终以Web界面呈现实时仪表板与统计结果同时借助NiFi完成数据可视化。目前已有53人学习适合作为大数据与物联网综合实践的参考案例便于快速搭建实验环境、理解各组件协作方式并开展二次开发。1. 智能家居数据从哪来、Spark 和 Kafka 各自扛什么活家里几十个传感器温湿度、门窗磁、人体红外、智能插座功率每秒都在往外吐数据。单机 Python 脚本收一条写一条数据库跑两天就卡成幻灯片——这不是数据量真有多大而是架构没搭对。基于 Spark 和 Kafka 的智能家居数据分析系统核心就干一件事让 Kafka 接住设备端的高频写入让 Spark 做流式清洗和批式聚合最后把结果落到能查的存储里。这套组合适合谁手上有树莓派或 ESP32 攒的传感器网络、想从「能亮灯」升级到「能看趋势、能告警」的开发者以及需要做数据分析项目但不想从零造轮子的人。Kafka 解决的是「收得下、不丢、能重放」Spark 解决的是「算得快、窗口灵活、批流一套代码」。两者拼在一起才撑得起一个能跑通、能演示、能继续往上加功能的智能家居数据分析系统。2. 消息队列选型与 Kafka 在智能家居场景的落地配置2.1 为什么智能家居数据管道优先选 Kafka 而不是 RabbitMQ热词里常出现「kafka、rabbitmq、rocketmq 消息队列选型实战对比与避坑指南」放到智能家居场景结论其实很干脆Kafka 的日志抽象天然适合传感器数据。RabbitMQ 强在路由灵活、消费确认精细但它的队列在消息堆积时性能下降明显而智能家居的典型负载是「平时每秒几十条有人在家时突然飙到几千条半夜又归零」——这种脉冲式写入RabbitMQ 的队列深度一上去内存和磁盘 IO 就开始报警。RocketMQ 事务消息和延迟消息做得漂亮但部署运维比 Kafka 重对小规模智能家居项目来说多出来的那套 NameServer 和 Broker 主从配置收益不明显。Kafka 的 Partition 机制让水平扩展变得直接设备数据按 device_id 哈希到不同 Partition消费者组里每个实例负责一部分吞吐不够就加 Partition、加消费者。更关键的是Kafka 的消息保留策略允许你按时间或大小保留原始数据智能家居场景里经常需要「回放昨天下午的温湿度曲线」Kafka 的 offset 重放比 RabbitMQ 的重新入队干净得多。我一般会跟团队说如果你需要的是「任务分发」选 RabbitMQ如果你需要的是「数据管道」选 Kafka。智能家居数据分析系统属于后者。2.2 单机与三节点 Kafka 集群的安装配置命令热词里「kafka 3节点集群 部署」「kafka安装配置」「kafka集群安装」出现频率很高说明很多人卡在环境搭建。先给单机开发环境的命令再补三节点集群的关键配置差异。单机 KafkaKRaft 模式不依赖 ZooKeeper适合本地开发# 下载并解压版本号按实际下载的填这里用 x.y.z 占位 tar -xzf kafka_2.13-x.y.z.tgz cd kafka_2.13-x.y.z # 生成集群 UUID KAFKA_CLUSTER_ID$(bin/kafka-storage.sh random-uuid) # 格式化日志目录KRaft 模式必须做 bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties # 启动 Broker bin/kafka-server-start.sh config/kraft/server.properties逻辑说明KRaft 模式下 Kafka 自己管理元数据不再需要单独起 ZooKeeper对智能家居这种中小规模场景少一个组件就少一个故障点。format命令只执行一次重复执行会清空数据。server.properties里需要确认listeners和advertised.listeners本机开发用PLAINTEXT://localhost:9092即可。三节点集群的关键差异在server.properties里这几项配置项node1node2node3说明node.id123每个节点唯一controller.quorum.voters1node1:9093,2node2:9093,3node3:9093同左同左KRaft 控制器投票成员listenersPLAINTEXT://:9092,CONTROLLER://:9093同左同左业务与控制器端口分开log.dirs/data/kafka-logs/data/kafka-logs/data/kafka-logs确保磁盘独立三节点部署时controller.quorum.voters三个节点必须写完全一致否则集群起不来。advertised.listeners要填其他节点能访问到的 IP 或主机名不能写 localhost这是新手翻车最多的地方。2.3 智能家居 Topic 设计与分区数估算Topic 设计直接决定后续 Spark 消费的并行度。我一般按数据类别拆 Topic而不是所有传感器塞一个# 创建传感器原始数据 Topic6 个分区副本因子 2三节点集群 bin/kafka-topics.sh --create \ --topic smart_home_sensor_raw \ --partitions 6 \ --replication-factor 2 \ --bootstrap-server node1:9092,node2:9092,node3:9092 # 创建告警事件 Topic3 个分区 bin/kafka-topics.sh --create \ --topic smart_home_alerts \ --partitions 3 \ --replication-factor 2 \ --bootstrap-server node1:9092,node2:9092,node3:9092分区数怎么估按峰值吞吐除以单分区可承载吞吐。单分区在普通机械盘上顺序写能到几十 MB/s但智能家居单条消息通常几百字节到 1KB瓶颈在消息条数而非字节数。经验值单分区每秒处理几千到一万条小消息没问题。如果峰值每秒 3 万条6 个分区留了余量。副本因子设 2 而不是 3是因为智能家居项目通常没有跨机房容灾需求2 副本已经能扛单节点故障同时省磁盘。注意分区数只能增不能减一开始别设太小但也别为了「以后可能用得上」设 100 个分区——每个分区都有文件句柄和内存开销Spark 消费时任务数也跟着涨。3. Spark 侧的数据接入、清洗与窗口聚合实现3.1 Spark 读取 Kafka 的依赖与最小可运行代码热词里「spark中读取json」「spark数据分析案例」「spark实战」指向很明确大家要的是能跑起来的代码。Spark 读 Kafka 需要spark-sql-kafka-0-10连接器版本要和 Spark 版本对齐。下面是最小可运行的 Structured Streaming 消费代码from pyspark.sql import SparkSession from pyspark.sql.functions import from_json, col, window, avg, max, min from pyspark.sql.types import StructType, StringType, DoubleType, LongType # 定义智能家居传感器数据的 JSON 结构 schema StructType() \ .add(device_id, StringType()) \ .add(room, StringType()) \ .add(metric, StringType()) \ .add(value, DoubleType()) \ .add(ts, LongType()) spark SparkSession.builder \ .appName(SmartHomeStreaming) \ .config(spark.sql.shuffle.partitions, 6) \ .getOrCreate() # 从 Kafka 读取原始流 raw_df spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, node1:9092,node2:9092,node3:9092) \ .option(subscribe, smart_home_sensor_raw) \ .option(startingOffsets, latest) \ .option(failOnDataLoss, false) \ .load() # 解析 JSON 并展开字段 parsed_df raw_df.select( from_json(col(value).cast(string), schema).alias(data) ).select(data.*) # 按房间和指标做 1 分钟滚动窗口聚合 agg_df parsed_df \ .withWatermark(ts, 2 minutes) \ .groupBy(window(col(ts), 1 minute), col(room), col(metric)) \ .agg(avg(value).alias(avg_val), max(value).alias(max_val), min(value).alias(min_val)) # 输出到控制台开发调试用 query agg_df.writeStream \ .outputMode(append) \ .format(console) \ .option(truncate, false) \ .start() query.awaitTermination()逻辑说明from_json把 Kafka 消息体里的 JSON 字符串按 schema 解析成列withWatermark设置 2 分钟水位线允许迟到 2 分钟内的数据参与窗口计算。window(col(ts), 1 minute)是滚动窗口不重叠。outputMode(append)配合水位线窗口在水位线超过窗口结束时间后才输出。参数说明startingOffsets设latest表示只消费启动后的新消息开发时常用生产环境如果要补历史数据改成earliest。failOnDataLoss设false避免因 offset 越界导致整个流挂掉但会丢消息生产环境要配合监控。spark.sql.shuffle.partitions默认 200小集群上设成和分区数相近的值减少小任务开销。3.2 数据清洗处理缺失值、异常值和重复上报智能家居传感器数据脏得很温湿度传感器偶尔吐 -999、门窗磁重复上报同一条状态、WiFi 抖动导致时间戳乱序。清洗逻辑要嵌在流里做from pyspark.sql.functions import when, lit, lag, abs as spark_abs from pyspark.sql.window import Window # 过滤明显异常值温度不在 -40~80湿度不在 0~100 cleaned_df parsed_df.filter( ((col(metric) temperature) (col(value).between(-40, 80))) | ((col(metric) humidity) (col(value).between(0, 100))) | ((col(metric) power) (col(value) 0)) ) # 去重同一设备同一指标同一秒只保留一条 dedup_df cleaned_df.dropDuplicates([device_id, metric, ts]) # 用前一条值填充缺失需要按设备指标分组按时间排序 window_spec Window.partitionBy(device_id, metric).orderBy(ts) filled_df dedup_df.withColumn( value_filled, when(col(value).isNull(), lag(value, 1).over(window_spec)).otherwise(col(value)) )逻辑说明异常值过滤用between做范围约束比写 UDF 快得多。dropDuplicates在流里会维护状态状态大小随去重键数量增长要配合水位线清理。lag填充需要窗口函数Structured Streaming 对窗口函数的支持有限如果流里做不了可以落到下游存储后用批处理补。参数说明温度范围 -40~80 覆盖了绝大多数室内外场景如果传感器装在特殊环境要调整。dropDuplicates的键选device_id metric tsts 是秒级时间戳同一秒内重复上报会被去掉。3.3 窗口聚合与告警规则从数据到可行动信息聚合结果要能回答「客厅温度过去 5 分钟是否持续高于 30 度」这类问题。滑动窗口比滚动窗口更适合告警# 5 分钟滑动窗口每 1 分钟滑动一次 sliding_agg parsed_df \ .withWatermark(ts, 3 minutes) \ .groupBy(window(col(ts), 5 minutes, 1 minute), col(room), col(metric)) \ .agg(avg(value).alias(avg_val), max(value).alias(max_val)) # 告警规则温度均值超过 30 或功率均值超过 2000W alert_df sliding_agg.filter( ((col(metric) temperature) (col(avg_val) 30)) | ((col(metric) power) (col(avg_val) 2000)) ).select( col(window.start).alias(window_start), col(window.end).alias(window_end), col(room), col(metric), col(avg_val) ) # 告警写入 Kafka 告警 Topic alert_query alert_df.writeStream \ .outputMode(append) \ .format(kafka) \ .option(kafka.bootstrap.servers, node1:9092,node2:9092,node3:9092) \ .option(topic, smart_home_alerts) \ .option(checkpointLocation, /data/checkpoints/alerts) \ .start()逻辑说明滑动窗口window(col(ts), 5 minutes, 1 minute)表示窗口长度 5 分钟、滑动步长 1 分钟同一个数据点会出现在 5 个窗口里适合做持续告警判断。checkpointLocation必须设否则每次重启流会从头消费或状态丢失。参数说明水位线设 3 分钟比窗口滑动步长 1 分钟大给迟到数据留了缓冲。告警阈值 30 度和 2000W 是示例值实际要按房间和设备类型调整——厨房温度和卧室温度不能用同一个阈值。4. 避坑与排查Kafka 消费延迟、Spark 状态膨胀和序列化翻车4.1 Kafka 消费延迟高先看消费者组和分区分配现象Spark 流启动后Kafka 的 consumer lag 持续上涨数据越积越多。原因通常有三个消费者数量少于分区数、单条消息处理逻辑太重、或者max.poll.records设太大导致一次拉太多处理不完。解决先确认消费者实例数是否等于分区数Structured Streaming 里一个 executor 核心对应一个分区消费任务如果分区 6 个但只给了 2 个核心剩下 4 个分区没人消费。再看maxOffsetsPerTrigger是否设了限制没设的话一次拉全量内存直接爆。我一般会设maxOffsetsPerTrigger为分区数乘以每分区每批期望处理条数比如 6 分区、每分区每批 5000 条就设 30000。4.2 Spark 状态膨胀导致 OOM水位线和去重键要一起调现象流跑几小时后 executor 内存告警日志里出现StateStore相关 OOM。原因dropDuplicates和窗口聚合都会维护状态如果水位线设得太大或者去重键基数太高状态永远不清理。解决水位线不要超过业务能容忍的迟到时间智能家居场景 2~3 分钟足够去重键避免用高基数字段比如别用ts毫秒级时间戳做去重键用秒级。另外spark.sql.streaming.stateStore.providerClass可以换成 RocksDB 实现状态存磁盘而不是堆内存代价是慢一点但不容易 OOM。4.3 JSON 解析失败静默丢数据加死信队列兜底现象Kafka 里消息数明明很多但 Spark 解析后输出条数对不上也没有报错。原因from_json遇到不符合 schema 的 JSON 会返回 null后续select(data.*)把 null 行也带过去了但字段全是 null聚合时被过滤掉看起来就像数据丢了。解决解析后加一步filter(col(data).isNotNull())同时把解析失败的消息写到死信 Topic# 解析失败的消息写到死信队列 bad_df raw_df.select( from_json(col(value).cast(string), schema).alias(data), col(value).alias(raw_value) ).filter(col(data).isNull()) bad_df.select(col(raw_value)) \ .writeStream \ .format(kafka) \ .option(kafka.bootstrap.servers, node1:9092,node2:9092,node3:9092) \ .option(topic, smart_home_dlq) \ .option(checkpointLocation, /data/checkpoints/dlq) \ .start()4.4 checkpoint 目录没设或设错重启后重复消费现象流任务重启后之前已经处理过的数据又被处理了一遍告警重复发送。原因checkpointLocation没设或者设在了临时目录被清理了或者多个流共用了同一个 checkpoint 目录。解决每个流必须有独立的 checkpoint 目录且放在持久化存储上。Structured Streaming 的 checkpoint 记录了 offset 和状态丢了就只能按startingOffsets重新来。我一般会在代码里把 checkpoint 路径写成hdfs://或挂载的持久卷路径本地开发用/data/checkpoints/并确保不会被tmpwatch清掉。4.5 时间戳用处理时间还是事件时间选错窗口全乱现象窗口聚合结果的时间段和实际数据时间对不上比如凌晨的数据被算到了早上。原因window函数默认按事件时间列但如果ts字段是处理时间Spark 收到消息的时间而不是设备上报时间窗口就错了。解决确保ts来自消息体里的设备时间戳并且是毫秒或秒级 Long 类型。如果设备时间不可信比如设备没联网对时那就只能用处理时间但要接受窗口边界不精确。智能家居场景我一般优先用设备时间戳同时在 Kafka 消息里保留一个ingest_ts用于排查。5. 把流式结果落到可查询存储从控制台到实际可用的最后一步开发时format(console)看输出没问题但真正要用起来得把聚合结果写到能查的地方。常见做法是写回 Kafka 再由下游消费或者直接写 Parquet 文件用 Spark SQL 查。我一般会双写告警走 Kafka 实时推送聚合指标写 Parquet 供历史查询。# 聚合结果写 Parquet按日期分区 parquet_query agg_df.writeStream \ .outputMode(append) \ .format(parquet) \ .option(path, /data/smart_home/agg) \ .option(checkpointLocation, /data/checkpoints/agg_parquet) \ .partitionBy(room, metric) \ .trigger(processingTime1 minute) \ .start()trigger(processingTime1 minute)表示每分钟触发一批而不是来一条算一条。智能家居数据分析对延迟要求通常到分钟级就够了微批模式比连续处理省资源得多。partitionBy按房间和指标分区后续查「客厅温度过去一周」时 Spark SQL 能直接做分区裁剪。验证方法起一个 Spark SQL 会话读 Parquet看窗口聚合结果是否和原始数据对得上。# 验证读 Parquet 查客厅温度均值 spark.read.parquet(/data/smart_home/agg) \ .filter((col(room) living_room) (col(metric) temperature)) \ .groupBy(window) \ .agg(avg(avg_val)) \ .orderBy(window) \ .show(20, truncateFalse)如果结果为空先检查 Parquet 目录下有没有_spark_metadata和实际数据文件再检查partitionBy的字段值是否和过滤条件一致——大小写、空格都可能导致查不到。另一个常见问题是 Parquet 文件太多每个微批产生一堆小文件需要定期 compaction或者用spark.sql.files.maxPartitionBytes控制读取时的合并。这套系统值不值得做如果你手上有智能家居设备数据想从「能看实时数值」升级到「能看趋势、能告警、能回溯」Kafka Spark 的组合是经过验证的路径。坑主要在环境配置和状态管理上但一旦跑通后续加设备、加指标、加告警规则都是改配置和 SQL 的事。我自己的习惯是先把单机 KRaft 跑通再用 Docker Compose 起三节点最后才上真实设备数据——跳过任何一步后面都要花更多时间补。希望帮到你。本文还有配套的精品资源点击获取 基于Spark与Kafka的智能家居数据分析系统实战
简介这是一套面向物联网与大数据方向学习者、开发者及课程设计者的智能家居数据分析系统源码围绕设备数据采集、流式传输、分布式存储与实时分析展开帮助读者理解从传感器到可视化仪表板的完整链路。资源包共16个文件包含zip依赖库、db数据库文件、yml容器编排配置、sql建表脚本、conf与env环境配置、ino设备端程序、py数据处理脚本及sh提交脚本等压缩包约174KB结构紧凑、模块划分清晰。项目以MQTT协议采集智能家居设备数据经Kafka消息队列保障实时性与可靠性由HDFS完成大规模存储再通过Spark进行高效处理分析并将结果写入PostgreSQL最终以Web界面呈现实时仪表板与统计结果同时借助NiFi完成数据可视化。目前已有53人学习适合作为大数据与物联网综合实践的参考案例便于快速搭建实验环境、理解各组件协作方式并开展二次开发。1. 智能家居数据从哪来、Spark 和 Kafka 各自扛什么活家里几十个传感器温湿度、门窗磁、人体红外、智能插座功率每秒都在往外吐数据。单机 Python 脚本收一条写一条数据库跑两天就卡成幻灯片——这不是数据量真有多大而是架构没搭对。基于 Spark 和 Kafka 的智能家居数据分析系统核心就干一件事让 Kafka 接住设备端的高频写入让 Spark 做流式清洗和批式聚合最后把结果落到能查的存储里。这套组合适合谁手上有树莓派或 ESP32 攒的传感器网络、想从「能亮灯」升级到「能看趋势、能告警」的开发者以及需要做数据分析项目但不想从零造轮子的人。Kafka 解决的是「收得下、不丢、能重放」Spark 解决的是「算得快、窗口灵活、批流一套代码」。两者拼在一起才撑得起一个能跑通、能演示、能继续往上加功能的智能家居数据分析系统。2. 消息队列选型与 Kafka 在智能家居场景的落地配置2.1 为什么智能家居数据管道优先选 Kafka 而不是 RabbitMQ热词里常出现「kafka、rabbitmq、rocketmq 消息队列选型实战对比与避坑指南」放到智能家居场景结论其实很干脆Kafka 的日志抽象天然适合传感器数据。RabbitMQ 强在路由灵活、消费确认精细但它的队列在消息堆积时性能下降明显而智能家居的典型负载是「平时每秒几十条有人在家时突然飙到几千条半夜又归零」——这种脉冲式写入RabbitMQ 的队列深度一上去内存和磁盘 IO 就开始报警。RocketMQ 事务消息和延迟消息做得漂亮但部署运维比 Kafka 重对小规模智能家居项目来说多出来的那套 NameServer 和 Broker 主从配置收益不明显。Kafka 的 Partition 机制让水平扩展变得直接设备数据按 device_id 哈希到不同 Partition消费者组里每个实例负责一部分吞吐不够就加 Partition、加消费者。更关键的是Kafka 的消息保留策略允许你按时间或大小保留原始数据智能家居场景里经常需要「回放昨天下午的温湿度曲线」Kafka 的 offset 重放比 RabbitMQ 的重新入队干净得多。我一般会跟团队说如果你需要的是「任务分发」选 RabbitMQ如果你需要的是「数据管道」选 Kafka。智能家居数据分析系统属于后者。2.2 单机与三节点 Kafka 集群的安装配置命令热词里「kafka 3节点集群 部署」「kafka安装配置」「kafka集群安装」出现频率很高说明很多人卡在环境搭建。先给单机开发环境的命令再补三节点集群的关键配置差异。单机 KafkaKRaft 模式不依赖 ZooKeeper适合本地开发# 下载并解压版本号按实际下载的填这里用 x.y.z 占位 tar -xzf kafka_2.13-x.y.z.tgz cd kafka_2.13-x.y.z # 生成集群 UUID KAFKA_CLUSTER_ID$(bin/kafka-storage.sh random-uuid) # 格式化日志目录KRaft 模式必须做 bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties # 启动 Broker bin/kafka-server-start.sh config/kraft/server.properties逻辑说明KRaft 模式下 Kafka 自己管理元数据不再需要单独起 ZooKeeper对智能家居这种中小规模场景少一个组件就少一个故障点。format命令只执行一次重复执行会清空数据。server.properties里需要确认listeners和advertised.listeners本机开发用PLAINTEXT://localhost:9092即可。三节点集群的关键差异在server.properties里这几项配置项node1node2node3说明node.id123每个节点唯一controller.quorum.voters1node1:9093,2node2:9093,3node3:9093同左同左KRaft 控制器投票成员listenersPLAINTEXT://:9092,CONTROLLER://:9093同左同左业务与控制器端口分开log.dirs/data/kafka-logs/data/kafka-logs/data/kafka-logs确保磁盘独立三节点部署时controller.quorum.voters三个节点必须写完全一致否则集群起不来。advertised.listeners要填其他节点能访问到的 IP 或主机名不能写 localhost这是新手翻车最多的地方。2.3 智能家居 Topic 设计与分区数估算Topic 设计直接决定后续 Spark 消费的并行度。我一般按数据类别拆 Topic而不是所有传感器塞一个# 创建传感器原始数据 Topic6 个分区副本因子 2三节点集群 bin/kafka-topics.sh --create \ --topic smart_home_sensor_raw \ --partitions 6 \ --replication-factor 2 \ --bootstrap-server node1:9092,node2:9092,node3:9092 # 创建告警事件 Topic3 个分区 bin/kafka-topics.sh --create \ --topic smart_home_alerts \ --partitions 3 \ --replication-factor 2 \ --bootstrap-server node1:9092,node2:9092,node3:9092分区数怎么估按峰值吞吐除以单分区可承载吞吐。单分区在普通机械盘上顺序写能到几十 MB/s但智能家居单条消息通常几百字节到 1KB瓶颈在消息条数而非字节数。经验值单分区每秒处理几千到一万条小消息没问题。如果峰值每秒 3 万条6 个分区留了余量。副本因子设 2 而不是 3是因为智能家居项目通常没有跨机房容灾需求2 副本已经能扛单节点故障同时省磁盘。注意分区数只能增不能减一开始别设太小但也别为了「以后可能用得上」设 100 个分区——每个分区都有文件句柄和内存开销Spark 消费时任务数也跟着涨。3. Spark 侧的数据接入、清洗与窗口聚合实现3.1 Spark 读取 Kafka 的依赖与最小可运行代码热词里「spark中读取json」「spark数据分析案例」「spark实战」指向很明确大家要的是能跑起来的代码。Spark 读 Kafka 需要spark-sql-kafka-0-10连接器版本要和 Spark 版本对齐。下面是最小可运行的 Structured Streaming 消费代码from pyspark.sql import SparkSession from pyspark.sql.functions import from_json, col, window, avg, max, min from pyspark.sql.types import StructType, StringType, DoubleType, LongType # 定义智能家居传感器数据的 JSON 结构 schema StructType() \ .add(device_id, StringType()) \ .add(room, StringType()) \ .add(metric, StringType()) \ .add(value, DoubleType()) \ .add(ts, LongType()) spark SparkSession.builder \ .appName(SmartHomeStreaming) \ .config(spark.sql.shuffle.partitions, 6) \ .getOrCreate() # 从 Kafka 读取原始流 raw_df spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, node1:9092,node2:9092,node3:9092) \ .option(subscribe, smart_home_sensor_raw) \ .option(startingOffsets, latest) \ .option(failOnDataLoss, false) \ .load() # 解析 JSON 并展开字段 parsed_df raw_df.select( from_json(col(value).cast(string), schema).alias(data) ).select(data.*) # 按房间和指标做 1 分钟滚动窗口聚合 agg_df parsed_df \ .withWatermark(ts, 2 minutes) \ .groupBy(window(col(ts), 1 minute), col(room), col(metric)) \ .agg(avg(value).alias(avg_val), max(value).alias(max_val), min(value).alias(min_val)) # 输出到控制台开发调试用 query agg_df.writeStream \ .outputMode(append) \ .format(console) \ .option(truncate, false) \ .start() query.awaitTermination()逻辑说明from_json把 Kafka 消息体里的 JSON 字符串按 schema 解析成列withWatermark设置 2 分钟水位线允许迟到 2 分钟内的数据参与窗口计算。window(col(ts), 1 minute)是滚动窗口不重叠。outputMode(append)配合水位线窗口在水位线超过窗口结束时间后才输出。参数说明startingOffsets设latest表示只消费启动后的新消息开发时常用生产环境如果要补历史数据改成earliest。failOnDataLoss设false避免因 offset 越界导致整个流挂掉但会丢消息生产环境要配合监控。spark.sql.shuffle.partitions默认 200小集群上设成和分区数相近的值减少小任务开销。3.2 数据清洗处理缺失值、异常值和重复上报智能家居传感器数据脏得很温湿度传感器偶尔吐 -999、门窗磁重复上报同一条状态、WiFi 抖动导致时间戳乱序。清洗逻辑要嵌在流里做from pyspark.sql.functions import when, lit, lag, abs as spark_abs from pyspark.sql.window import Window # 过滤明显异常值温度不在 -40~80湿度不在 0~100 cleaned_df parsed_df.filter( ((col(metric) temperature) (col(value).between(-40, 80))) | ((col(metric) humidity) (col(value).between(0, 100))) | ((col(metric) power) (col(value) 0)) ) # 去重同一设备同一指标同一秒只保留一条 dedup_df cleaned_df.dropDuplicates([device_id, metric, ts]) # 用前一条值填充缺失需要按设备指标分组按时间排序 window_spec Window.partitionBy(device_id, metric).orderBy(ts) filled_df dedup_df.withColumn( value_filled, when(col(value).isNull(), lag(value, 1).over(window_spec)).otherwise(col(value)) )逻辑说明异常值过滤用between做范围约束比写 UDF 快得多。dropDuplicates在流里会维护状态状态大小随去重键数量增长要配合水位线清理。lag填充需要窗口函数Structured Streaming 对窗口函数的支持有限如果流里做不了可以落到下游存储后用批处理补。参数说明温度范围 -40~80 覆盖了绝大多数室内外场景如果传感器装在特殊环境要调整。dropDuplicates的键选device_id metric tsts 是秒级时间戳同一秒内重复上报会被去掉。3.3 窗口聚合与告警规则从数据到可行动信息聚合结果要能回答「客厅温度过去 5 分钟是否持续高于 30 度」这类问题。滑动窗口比滚动窗口更适合告警# 5 分钟滑动窗口每 1 分钟滑动一次 sliding_agg parsed_df \ .withWatermark(ts, 3 minutes) \ .groupBy(window(col(ts), 5 minutes, 1 minute), col(room), col(metric)) \ .agg(avg(value).alias(avg_val), max(value).alias(max_val)) # 告警规则温度均值超过 30 或功率均值超过 2000W alert_df sliding_agg.filter( ((col(metric) temperature) (col(avg_val) 30)) | ((col(metric) power) (col(avg_val) 2000)) ).select( col(window.start).alias(window_start), col(window.end).alias(window_end), col(room), col(metric), col(avg_val) ) # 告警写入 Kafka 告警 Topic alert_query alert_df.writeStream \ .outputMode(append) \ .format(kafka) \ .option(kafka.bootstrap.servers, node1:9092,node2:9092,node3:9092) \ .option(topic, smart_home_alerts) \ .option(checkpointLocation, /data/checkpoints/alerts) \ .start()逻辑说明滑动窗口window(col(ts), 5 minutes, 1 minute)表示窗口长度 5 分钟、滑动步长 1 分钟同一个数据点会出现在 5 个窗口里适合做持续告警判断。checkpointLocation必须设否则每次重启流会从头消费或状态丢失。参数说明水位线设 3 分钟比窗口滑动步长 1 分钟大给迟到数据留了缓冲。告警阈值 30 度和 2000W 是示例值实际要按房间和设备类型调整——厨房温度和卧室温度不能用同一个阈值。4. 避坑与排查Kafka 消费延迟、Spark 状态膨胀和序列化翻车4.1 Kafka 消费延迟高先看消费者组和分区分配现象Spark 流启动后Kafka 的 consumer lag 持续上涨数据越积越多。原因通常有三个消费者数量少于分区数、单条消息处理逻辑太重、或者max.poll.records设太大导致一次拉太多处理不完。解决先确认消费者实例数是否等于分区数Structured Streaming 里一个 executor 核心对应一个分区消费任务如果分区 6 个但只给了 2 个核心剩下 4 个分区没人消费。再看maxOffsetsPerTrigger是否设了限制没设的话一次拉全量内存直接爆。我一般会设maxOffsetsPerTrigger为分区数乘以每分区每批期望处理条数比如 6 分区、每分区每批 5000 条就设 30000。4.2 Spark 状态膨胀导致 OOM水位线和去重键要一起调现象流跑几小时后 executor 内存告警日志里出现StateStore相关 OOM。原因dropDuplicates和窗口聚合都会维护状态如果水位线设得太大或者去重键基数太高状态永远不清理。解决水位线不要超过业务能容忍的迟到时间智能家居场景 2~3 分钟足够去重键避免用高基数字段比如别用ts毫秒级时间戳做去重键用秒级。另外spark.sql.streaming.stateStore.providerClass可以换成 RocksDB 实现状态存磁盘而不是堆内存代价是慢一点但不容易 OOM。4.3 JSON 解析失败静默丢数据加死信队列兜底现象Kafka 里消息数明明很多但 Spark 解析后输出条数对不上也没有报错。原因from_json遇到不符合 schema 的 JSON 会返回 null后续select(data.*)把 null 行也带过去了但字段全是 null聚合时被过滤掉看起来就像数据丢了。解决解析后加一步filter(col(data).isNotNull())同时把解析失败的消息写到死信 Topic# 解析失败的消息写到死信队列 bad_df raw_df.select( from_json(col(value).cast(string), schema).alias(data), col(value).alias(raw_value) ).filter(col(data).isNull()) bad_df.select(col(raw_value)) \ .writeStream \ .format(kafka) \ .option(kafka.bootstrap.servers, node1:9092,node2:9092,node3:9092) \ .option(topic, smart_home_dlq) \ .option(checkpointLocation, /data/checkpoints/dlq) \ .start()4.4 checkpoint 目录没设或设错重启后重复消费现象流任务重启后之前已经处理过的数据又被处理了一遍告警重复发送。原因checkpointLocation没设或者设在了临时目录被清理了或者多个流共用了同一个 checkpoint 目录。解决每个流必须有独立的 checkpoint 目录且放在持久化存储上。Structured Streaming 的 checkpoint 记录了 offset 和状态丢了就只能按startingOffsets重新来。我一般会在代码里把 checkpoint 路径写成hdfs://或挂载的持久卷路径本地开发用/data/checkpoints/并确保不会被tmpwatch清掉。4.5 时间戳用处理时间还是事件时间选错窗口全乱现象窗口聚合结果的时间段和实际数据时间对不上比如凌晨的数据被算到了早上。原因window函数默认按事件时间列但如果ts字段是处理时间Spark 收到消息的时间而不是设备上报时间窗口就错了。解决确保ts来自消息体里的设备时间戳并且是毫秒或秒级 Long 类型。如果设备时间不可信比如设备没联网对时那就只能用处理时间但要接受窗口边界不精确。智能家居场景我一般优先用设备时间戳同时在 Kafka 消息里保留一个ingest_ts用于排查。5. 把流式结果落到可查询存储从控制台到实际可用的最后一步开发时format(console)看输出没问题但真正要用起来得把聚合结果写到能查的地方。常见做法是写回 Kafka 再由下游消费或者直接写 Parquet 文件用 Spark SQL 查。我一般会双写告警走 Kafka 实时推送聚合指标写 Parquet 供历史查询。# 聚合结果写 Parquet按日期分区 parquet_query agg_df.writeStream \ .outputMode(append) \ .format(parquet) \ .option(path, /data/smart_home/agg) \ .option(checkpointLocation, /data/checkpoints/agg_parquet) \ .partitionBy(room, metric) \ .trigger(processingTime1 minute) \ .start()trigger(processingTime1 minute)表示每分钟触发一批而不是来一条算一条。智能家居数据分析对延迟要求通常到分钟级就够了微批模式比连续处理省资源得多。partitionBy按房间和指标分区后续查「客厅温度过去一周」时 Spark SQL 能直接做分区裁剪。验证方法起一个 Spark SQL 会话读 Parquet看窗口聚合结果是否和原始数据对得上。# 验证读 Parquet 查客厅温度均值 spark.read.parquet(/data/smart_home/agg) \ .filter((col(room) living_room) (col(metric) temperature)) \ .groupBy(window) \ .agg(avg(avg_val)) \ .orderBy(window) \ .show(20, truncateFalse)如果结果为空先检查 Parquet 目录下有没有_spark_metadata和实际数据文件再检查partitionBy的字段值是否和过滤条件一致——大小写、空格都可能导致查不到。另一个常见问题是 Parquet 文件太多每个微批产生一堆小文件需要定期 compaction或者用spark.sql.files.maxPartitionBytes控制读取时的合并。这套系统值不值得做如果你手上有智能家居设备数据想从「能看实时数值」升级到「能看趋势、能告警、能回溯」Kafka Spark 的组合是经过验证的路径。坑主要在环境配置和状态管理上但一旦跑通后续加设备、加指标、加告警规则都是改配置和 SQL 的事。我自己的习惯是先把单机 KRaft 跑通再用 Docker Compose 起三节点最后才上真实设备数据——跳过任何一步后面都要花更多时间补。希望帮到你。本文还有配套的精品资源点击获取