大家好我是Java1234_小鋒老師分享一套鋒哥原創(chuàng)的基于Spark實(shí)時(shí)物聯(lián)網(wǎng)設(shè)備故障預(yù)警 數(shù)據(jù)分析與預(yù)測(cè) 系統(tǒng)(Java版本可視化大屏KafkaSpringBootVue3)項(xiàng)目介紹隨著工業(yè)互聯(lián)網(wǎng)與智能制造的快速發(fā)展物聯(lián)網(wǎng)設(shè)備在生產(chǎn)現(xiàn)場(chǎng)的部署規(guī)模持續(xù)擴(kuò)大。傳統(tǒng)依賴人工巡檢與閾值告警的運(yùn)維方式難以應(yīng)對(duì)高頻、多維、持續(xù)到達(dá)的設(shè)備傳感器數(shù)據(jù)容易出現(xiàn)故障發(fā)現(xiàn)滯后、誤報(bào)漏報(bào)較多、運(yùn)維成本居高不下等問題。針對(duì)上述背景本文設(shè)計(jì)并實(shí)現(xiàn)了一套基于 Spark 思想的實(shí)時(shí)物聯(lián)網(wǎng)設(shè)備故障預(yù)警系統(tǒng)綜合運(yùn)用 Java、Spring Boot、Vue3、Kafka、MySQL 等主流技術(shù)完成了從數(shù)據(jù)采集、實(shí)時(shí)分析、風(fēng)險(xiǎn)評(píng)分、故障預(yù)警到可視化展示與風(fēng)險(xiǎn)預(yù)測(cè)的完整閉環(huán)。系統(tǒng)后端采用 Spring Boot 構(gòu)建 RESTful 服務(wù)結(jié)合 Spring Security 與 JWT 實(shí)現(xiàn)管理員身份認(rèn)證與接口鑒權(quán)前端采用 Vue3、Element Plus 與 ECharts 實(shí)現(xiàn)管理后臺(tái)與數(shù)據(jù)可視化大屏數(shù)據(jù)采集側(cè)通過 Kafka 消息主題device_telemetry 緩沖設(shè)備遙測(cè)數(shù)據(jù)并在 Kafka 不可用時(shí)自動(dòng)降級(jí)為純 Java 寫庫保證演示與實(shí)驗(yàn)環(huán)境的可用性。系統(tǒng)以窗口聚合方式統(tǒng)計(jì)設(shè)備數(shù)量、告警數(shù)量、平均溫度、平均振動(dòng)、故障率與風(fēng)險(xiǎn)評(píng)分基于溫度、振動(dòng)、電流等指標(biāo)計(jì)算設(shè)備故障風(fēng)險(xiǎn)并利用多元線性回歸對(duì)風(fēng)險(xiǎn)評(píng)分序列進(jìn)行預(yù)測(cè)輸出 RMSE、MAE、MAPE 等誤差指標(biāo)。數(shù)據(jù)庫采用 MySQL庫名為 db_iot_fault核心數(shù)據(jù)表包括管理員表、設(shè)備類型表、設(shè)備表、傳感器數(shù)據(jù)表、故障預(yù)警表、實(shí)時(shí)統(tǒng)計(jì)表、預(yù)測(cè)結(jié)果表與誤差指標(biāo)表。系統(tǒng)實(shí)現(xiàn)了登錄認(rèn)證、首頁統(tǒng)計(jì)、設(shè)備管理、傳感器數(shù)據(jù)查詢、故障預(yù)警處理、實(shí)時(shí)分析、預(yù)測(cè)分析、可視化大屏以及個(gè)人中心資料修改、頭像上傳、密碼修改等功能。測(cè)試結(jié)果表明系統(tǒng)能夠穩(wěn)定完成實(shí)時(shí)數(shù)據(jù)采集、預(yù)警生成與預(yù)測(cè)分析界面交互流暢滿足本科畢業(yè)設(shè)計(jì)對(duì)完整性、實(shí)用性與技術(shù)綜合性的要求。源碼下載鏈接: https://pan.baidu.com/s/1YtIlK_Xw-u7Z8Qu_Mr16Bw?pwd1234提取碼: 1234系統(tǒng)展示核心代碼package com.java1234.spark; import com.java1234.config.AppProperties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; import org.springframework.stereotype.Component; import java.util.List; import java.util.Properties; /** * Kafka 遙測(cè)生產(chǎn)者可選失敗時(shí)降級(jí) */ Component public class KafkaTelemetryProducer { private final AppProperties appProperties; public KafkaTelemetryProducer(AppProperties appProperties) { this.appProperties appProperties; } /** * 嘗試發(fā)送事件到 Kafka */ public boolean trySend(ListTelemetryEvent events) { Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 127.0.0.1:9092); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 3000); props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 5000); props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 500); props.put(ProducerConfig.RETRIES_CONFIG, 0); try (KafkaProducerString, String producer new KafkaProducer(props)) { for (TelemetryEvent event : events) { String json toJson(event); producer.send(new ProducerRecord(appProperties.getKafkaTopic(), json)).get(); } return true; } catch (Exception ex) { System.out.println([Simulator] Kafka 發(fā)送失敗使用純 Java 寫庫: ex.getMessage()); return false; } } /** * 將遙測(cè)事件序列化為 JSON */ private String toJson(TelemetryEvent event) { return String.format( {\device_id\:%d,\temperature\:%s,\vibration\:%s,\current\:%s,\voltage\:%s,\humidity\:%s,\is_fault\:%d,\risk_score\:%s,\event_time\:\%s\}, event.getDeviceId(), event.getTemperature(), event.getVibration(), event.getCurrent(), event.getVoltage(), event.getHumidity(), event.getIsFault(), event.getRiskScore(), event.getEventTime().format(java.time.format.DateTimeFormatter.ofPattern(yyyy-MM-dd HH:mm:ss)) ); } }template div classpage-container div classpage-card div classpage-title故障風(fēng)險(xiǎn)預(yù)測(cè)分析/div div classerror-cards div classerror-carddiv classmetric-labelRMSE (均方根誤差)/divdiv classmetric-value{{ errorMetric.rmse }}/div/div div classerror-carddiv classmetric-labelMAE (平均絕對(duì)誤差)/divdiv classmetric-value{{ errorMetric.mae }}/div/div div classerror-carddiv classmetric-labelMAPE (平均絕對(duì)百分比誤差 %)/divdiv classmetric-value{{ errorMetric.mape }}%/div/div /div div refcompareRef classpred-chart/div div refresidualRef classpred-chart pred-residual/div el-table :datatableData stripe border el-table-column propwindow_time label時(shí)間窗口 min-width170 template #default{ row }{{ formatWindowTime(row.window_time) }}/template /el-table-column el-table-column proptrue_score label真實(shí)風(fēng)險(xiǎn)分 min-width120 template #default{ row }span stylecolor:#409eff;font-weight:600{{ row.true_score }}/span/template /el-table-column el-table-column proppred_score label預(yù)測(cè)風(fēng)險(xiǎn)分 min-width120 template #default{ row }span stylecolor:#67c23a;font-weight:600{{ row.pred_score }}/span/template /el-table-column el-table-column label誤差 min-width100 template #default{ row } span :style{ color: Math.abs(row.true_score - row.pred_score) 5 ? #f56c6c : #909399 } {{ (row.true_score - row.pred_score).toFixed(2) }} /span /template /el-table-column el-table-column propcreate_time label生成時(shí)間 min-width170 template #default{ row }{{ formatDateTime(row.create_time) }}/template /el-table-column /el-table el-pagination stylemargin-top:16px;justify-content:flex-end v-model:current-pagepage v-model:page-sizesize :totaltotal layouttotal, prev, pager, next changeloadTable / /div /div /template script setup /** * 預(yù)測(cè)分析頁面真實(shí) vs 預(yù)測(cè)對(duì)比 誤差分析 */ import { ref, onMounted, onUnmounted } from vue import * as echarts from echarts import request from /utils/request import { formatDateTime, formatWindowTime } from /utils/format const errorMetric ref({ rmse: 0, mae: 0, mape: 0 }) const tableData ref([]) const page ref(1) const size ref(10) const total ref(0) const compareRef ref(null) const residualRef ref(null) let charts [] let pollTimer null function axisLabel() { return { rotate: 30, interval: auto, formatter(val) { const t formatWindowTime(val); return t.length 16 ? ${t.slice(0,10)}\n${t.slice(11)} : t } } } function initCompareChart(data) { if (!compareRef.value) return const chart echarts.init(compareRef.value) const labels data.map(d formatWindowTime(d.window_time)) chart.setOption({ title: { text: 真實(shí)風(fēng)險(xiǎn)分 vs 預(yù)測(cè)風(fēng)險(xiǎn)分 對(duì)比, left: center, textStyle: { fontSize: 15 } }, tooltip: { trigger: axis }, legend: { data: [真實(shí)風(fēng)險(xiǎn)分, 預(yù)測(cè)風(fēng)險(xiǎn)分], top: 32 }, grid: { left: 20, right: 24, bottom: 28, top: 72, containLabel: true }, xAxis: { type: category, data: labels, axisLabel: axisLabel() }, yAxis: { type: value, name: 風(fēng)險(xiǎn)評(píng)分 }, series: [ { name: 真實(shí)風(fēng)險(xiǎn)分, type: line, smooth: true, data: data.map(d Number(d.true_score)), itemStyle: { color: #409eff }, lineStyle: { width: 3 } }, { name: 預(yù)測(cè)風(fēng)險(xiǎn)分, type: line, smooth: true, data: data.map(d Number(d.pred_score)), itemStyle: { color: #67c23a }, lineStyle: { width: 3, type: dashed } }, ], }) charts.push(chart) } function initResidualChart(data) { if (!residualRef.value) return const chart echarts.init(residualRef.value) const labels data.map(d formatWindowTime(d.window_time)) const residuals data.map(d Number((Number(d.true_score) - Number(d.pred_score)).toFixed(2))) chart.setOption({ title: { text: 預(yù)測(cè)殘差分析 (真實(shí)值 - 預(yù)測(cè)值), left: center, textStyle: { fontSize: 15 } }, tooltip: { trigger: axis }, grid: { left: 20, right: 24, bottom: 28, top: 56, containLabel: true }, xAxis: { type: category, data: labels, axisLabel: axisLabel() }, yAxis: { type: value, name: 殘差 }, series: [{ type: bar, data: residuals.map(v ({ value: v, itemStyle: { color: v 0 ? #409eff : #f56c6c } })), barWidth: 20 }], }) charts.push(chart) } async function loadData() { const [errorRes, compareRes] await Promise.all([ request.get(/prediction/error), request.get(/prediction/compare), ]) errorMetric.value errorRes.data charts.forEach(c c.dispose()) charts [] initCompareChart(compareRes.data || []) initResidualChart(compareRes.data || []) } async function loadTable() { const res await request.get(/prediction/list, { params: { page: page.value, size: size.value } }) tableData.value res.data.items total.value res.data.total } onMounted(() { loadData() loadTable() pollTimer setInterval(loadData, 5000) }) onUnmounted(() { clearInterval(pollTimer); charts.forEach(c c.dispose()) }) /script style scoped .pred-chart { width: 100%; height: 420px; margin-bottom: 24px; } .pred-residual { height: 360px; } /style