【免费】基于Spark实时物联网设备故障预警 数据分析与预测 系统(Python版本+pyspark+可视化大屏+Kafka+FastAPI+Vue3) 锋哥原创出品，必属精品

大家好，我是Java1234_小锋老师，分享一套锋哥原创的基于Spark实时物联网设备故障预警 数据分析与预测 系统(Python版本+pyspark+可视化大屏+Kafka+FastAPI+Vue3)

项目介绍

随着工业互联网与智能制造的深入推进，工厂现场部署了大量物联网传感设备。这些设备在运行过程中持续产生温度、振动、电流、电压、湿度等高频遥测数据。若仍采用传统的人工巡检或离线批处理分析方式，难以及时发现设备劣化趋势，容易造成非计划停机、产品质量波动甚至安全事故。因此，构建一套能够对设备状态进行实时感知、快速分析与风险预警的系统，具有明确的工程应用价值和教学研究意义。

本文设计并实现了“基于Spark实时物联网设备故障预警系统”。系统采用前后端分离架构：前端基于 Vue3、Element Plus 与 ECharts 构建管理后台与可视化数据大屏；后端采用 Python 语言与 FastAPI 框架提供 REST 接口，完成认证鉴权、设备管理、传感器查询、预警处理、个人中心等业务；数据采集与传输环节引入 Kafka 作为高吞吐消息中间件；实时计算环节基于 Spark Streaming（Structured Streaming）完成窗口聚合、异常清洗与风险统计；预测环节结合 Spark ML 线性回归模型对风险评分进行建模预测，并输出 RMSE、MAE、MAPE 等误差指标。持久化存储采用 MySQL，数据库命名为 db_iot_fault，业务表均以 t_ 前缀规范命名。

系统实现了管理员登录与个人中心、设备类型与设备档案管理、传感器历史数据查询、故障预警处理、实时统计分析、风险预测评估以及数据大屏展示等功能。通过模拟物联网设备数据流验证了从采集、传输、计算到展示的完整闭环。测试结果表明，系统能够稳定完成实时统计与预警推送，预测模块可对风险走势给出可解释的量化评估，满足本科毕业设计对完整性、可用性与技术综合性的要求。

源码下载 

链接: https://pan.baidu.com/s/1roSBANb7IpQZxgxAtUq0lw?pwd=1234 提取码: 1234

系统展示

核心代码

"""
预测分析路由
"""
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session

from database import get_db
from models.prediction import Prediction
from models.error_metric import ErrorMetric
from schemas import ResponseModel, PageResult, PredictionOut, ErrorMetricOut

router = APIRouter(prefix="/prediction", tags=["预测分析"])

def _normalize_window_time(wt: str) -> str:
"""将窗口时间规范为完整日期时间显示"""
if not wt:
return ""
text = str(wt).strip()
if len(text) == 13 and text[10] == " ":
return f"{text}:00:00"
if len(text) == 16 and text[13] == ":":
return f"{text}:00"
return text

def _to_prediction_out(row: Prediction) -> PredictionOut:
"""转换为输出模型"""
item = PredictionOut.model_validate(row)
item.window_time = _normalize_window_time(item.window_time)
return item

@router.get("/list", response_model=ResponseModel[PageResult[PredictionOut]])
def list_predictions(
page: int = Query(1, ge=1),
size: int = Query(10, ge=1, le=100),
db: Session = Depends(get_db),
):
"""
分页查询预测结果
"""
total = db.query(Prediction).count()
rows = (
db.query(Prediction)
.order_by(Prediction.window_time.asc())
.offset((page – 1) * size)
.limit(size)
.all()
)
items = [_to_prediction_out(r) for r in rows]
return ResponseModel(data=PageResult(total=total, items=items))

@router.get("/compare", response_model=ResponseModel[list[PredictionOut]])
def get_compare_chart(db: Session = Depends(get_db)):
"""
获取真实值 vs 预测值对比数据
"""
rows = db.query(Prediction).order_by(Prediction.window_time.asc()).all()
return ResponseModel(data=[_to_prediction_out(r) for r in rows])

@router.get("/error", response_model=ResponseModel[ErrorMetricOut])
def get_error_metric(db: Session = Depends(get_db)):
"""
获取最新误差指标
"""
metric = (
db.query(ErrorMetric)
.order_by(ErrorMetric.create_time.desc())
.first()
)
if not metric:
return ResponseModel(data=ErrorMetricOut(id=0, rmse=0, mae=0, mape=0))
return ResponseModel(data=ErrorMetricOut.model_validate(metric))

<template>
<div class="page-container">
<div class="page-card">
<div class="page-title">设备管理</div>
<div class="toolbar">
<el-input v-model="keyword" placeholder="搜索设备名称/编号" clearable style="width:220px" @keyup.enter="loadData" />
<el-button type="primary" @click="loadData">查询</el-button>
<el-button type="success" @click="openDialog()">新增设备</el-button>
</div>
<el-table :data="tableData" stripe border>
<el-table-column prop="device_code" label="设备编号" min-width="120" />
<el-table-column prop="name" label="设备名称" min-width="140" />
<el-table-column prop="type_name" label="设备类型" min-width="110" />
<el-table-column prop="location" label="安装位置" min-width="130" />
<el-table-column prop="status" label="状态" min-width="90">
<template #default="{ row }">
<el-tag :type="statusType(row.status)">{{ statusText(row.status) }}</el-tag>
</template>
</el-table-column>
<el-table-column prop="create_time" label="创建时间" min-width="170">
<template #default="{ row }">{{ formatDateTime(row.create_time) }}</template>
</el-table-column>
<el-table-column label="操作" min-width="160" fixed="right">
<template #default="{ row }">
<el-button link type="primary" @click="openDialog(row)">编辑</el-button>
<el-button link type="danger" @click="handleDelete(row)">删除</el-button>
</template>
</el-table-column>
</el-table>
<el-pagination style="margin-top:16px;justify-content:flex-end" v-model:current-page="page" v-model:page-size="size" :total="total" layout="total, prev, pager, next" @change="loadData" />
</div>

<el-dialog v-model="dialogVisible" :title="form.id ? '编辑设备' : '新增设备'" width="520px">
<el-form :model="form" label-width="90px">
<el-form-item label="设备编号"><el-input v-model="form.device_code" /></el-form-item>
<el-form-item label="设备名称"><el-input v-model="form.name" /></el-form-item>
<el-form-item label="设备类型">
<el-select v-model="form.type_id" style="width:100%">
<el-option v-for="t in types" :key="t.id" :label="t.name" :value="t.id" />
</el-select>
</el-form-item>
<el-form-item label="安装位置"><el-input v-model="form.location" /></el-form-item>
<el-form-item label="状态">
<el-select v-model="form.status" style="width:100%">
<el-option label="在线" value="online" />
<el-option label="离线" value="offline" />
<el-option label="故障" value="fault" />
</el-select>
</el-form-item>
</el-form>
<template #footer>
<el-button @click="dialogVisible = false">取消</el-button>
<el-button type="primary" @click="saveDevice">保存</el-button>
</template>
</el-dialog>
</div>
</template>

<script setup>
/**
* 设备管理页面
*/
import { ref, reactive, onMounted } from 'vue'
import request from '@/utils/request'
import { formatDateTime } from '@/utils/format'
import { ElMessage, ElMessageBox } from 'element-plus'

const tableData = ref([])
const types = ref([])
const keyword = ref('')
const page = ref(1)
const size = ref(10)
const total = ref(0)
const dialogVisible = ref(false)
const form = reactive({ id: null, device_code: '', name: '', type_id: null, location: '', status: 'online' })

function statusType(s) { return { online: 'success', offline: 'info', fault: 'danger' }[s] || 'info' }
function statusText(s) { return { online: '在线', offline: '离线', fault: '故障' }[s] || s }

async function loadTypes() {
const res = await request.get('/device/types')
types.value = res.data
}

async function loadData() {
const res = await request.get('/device/list', { params: { page: page.value, size: size.value, keyword: keyword.value } })
tableData.value = res.data.items
total.value = res.data.total
}

function openDialog(row) {
if (row) Object.assign(form, { id: row.id, device_code: row.device_code, name: row.name, type_id: row.type_id, location: row.location, status: row.status })
else Object.assign(form, { id: null, device_code: '', name: '', type_id: types.value[0]?.id, location: '', status: 'online' })
dialogVisible.value = true
}

async function saveDevice() {
const payload = { device_code: form.device_code, name: form.name, type_id: form.type_id, location: form.location, status: form.status }
if (form.id) await request.put(`/device/update/${form.id}`, payload)
else await request.post('/device/create', payload)
ElMessage.success('保存成功')
dialogVisible.value = false
loadData()
}

async function handleDelete(row) {
await ElMessageBox.confirm('确定删除该设备吗？', '提示', { type: 'warning' })
await request.delete(`/device/delete/${row.id}`)
ElMessage.success('删除成功')
loadData()
}

onMounted(() => { loadTypes(); loadData() })
</script>