 锋哥原创出品，必属精品)
大家好我是Java1234_小锋老师分享一套锋哥原创的基于Spark实时电商用户行为分析与预测 系统(Python版本pyspark可视化大屏KafkaFastAPIVue3)项目介绍随着电子商务规模持续扩大用户在浏览、加购、收藏与购买等环节产生的行为数据呈现高并发、高吞吐与强时效特征。传统离线批处理分析难以满足运营决策对实时性的要求。本文设计并实现了一套基于 Spark 的实时电商用户行为分析与预测系统围绕“数据采集—流式计算—指标落库—可视化展示—销售预测”的完整链路展开研究与工程实践。系统采用前后端分离架构前端基于 Vue3、Element Plus 与 ECharts 构建管理端与数据大屏后端采用 Python FastAPI 提供 RESTful 接口并结合 JWT 完成管理员身份认证实时链路以 Kafka 作为消息中间件承接行为事件以 Spark Structured Streaming 完成按小时窗口的 PV、UV、加购、收藏、购买与销售额聚合预测模块基于 Spark ML 线性回归对销售额序列进行建模并输出 RMSE、MAE、MAPE 等误差指标。数据持久化采用 MySQL数据库名为 db_ecommerce核心业务表均以 t_ 前缀命名。测试结果表明系统能够稳定完成管理员登录、个人中心维护、行为与商品管理、实时统计展示、销售预测对比及流水线状态监控等功能具备较好的可扩展性与教学示范价值可为电商运营提供实时洞察与辅助决策支持。本文的主要工作包括完成系统需求分析与总体架构设计绘制实体属性图与实体关系图并完成八张核心业务表设计实现基于 Kafka 与 Spark 的实时统计及销售预测链路完成 Vue3 管理端与数据大屏开展功能测试并给出改进方向。研究结果表明将流式计算与 Web 管理系统结合能够在本科毕业设计条件下形成完整、可运行、可解释的实时分析应用。源码下载链接: https://pan.baidu.com/s/1u0yzt7SCx13nEjaH4FSMLw?pwd1234提取码: 1234系统展示核心代码 Spark ML 销售额预测模块 import numpy as np from decimal import Decimal from config import settings def compute_error_metrics(y_true: list, y_pred: list) - dict: 计算误差指标RMSE、MAE、MAPE y_true np.array(y_true, dtypefloat) y_pred np.array(y_pred, dtypefloat) rmse float(np.sqrt(np.mean((y_true - y_pred) ** 2))) mae float(np.mean(np.abs(y_true - y_pred))) mask y_true ! 0 if mask.any(): mape float(np.mean(np.abs((y_true[mask] - y_pred[mask]) / y_true[mask])) * 100) else: mape 0.0 return {rmse: round(rmse, 4), mae: round(mae, 4), mape: round(mape, 4)} def run_spark_prediction(sales_series: list None) - tuple: 使用 Spark ML 进行销售额预测 返回 (predictions, error_metrics) try: from pyspark.sql import SparkSession from pyspark.ml.feature import VectorAssembler from pyspark.ml.regression import LinearRegression from pyspark.sql.types import StructType, StructField, DoubleType, IntegerType, StringType import pyspark.sql.functions as F spark SparkSession.builder \ .appName(SalesPrediction) \ .master(settings.SPARK_MASTER) \ .config(spark.driver.memory, 2g) \ .getOrCreate() spark.sparkContext.setLogLevel(WARN) if sales_series is None: from database import SessionLocal from models.realtime_stat import RealtimeStat db SessionLocal() stats db.query(RealtimeStat).order_by(RealtimeStat.window_time.asc()).all() db.close() sales_series [ {window_time: s.window_time, sales: float(s.sales)} for s in stats ] if len(sales_series) 5: spark.stop() return [], {rmse: 0, mae: 0, mape: 0} # 构造滞后特征 data [] for i in range(3, len(sales_series)): data.append({ window_time: sales_series[i][window_time], lag1: sales_series[i - 1][sales], lag2: sales_series[i - 2][sales], lag3: sales_series[i - 3][sales], hour: int(str(sales_series[i][window_time])[11:13]), sales: sales_series[i][sales], }) schema StructType([ StructField(window_time, StringType()), StructField(lag1, DoubleType()), StructField(lag2, DoubleType()), StructField(lag3, DoubleType()), StructField(hour, IntegerType()), StructField(sales, DoubleType()), ]) df spark.createDataFrame(data, schema) # 划分训练集和测试集后20%作为测试 split_idx max(int(len(data) * 0.8), 1) train_df df.limit(split_idx) test_df df.filter(F.monotonically_increasing_id() split_idx) assembler VectorAssembler( inputCols[lag1, lag2, lag3, hour], outputColfeatures ) train_df assembler.transform(train_df) test_df assembler.transform(test_df) lr LinearRegression(featuresColfeatures, labelColsales, maxIter100) model lr.fit(train_df) predictions_raw model.transform(test_df).collect() predictions [] y_true, y_pred [], [] for row in predictions_raw: true_val float(row[sales]) pred_val float(row[prediction]) predictions.append({ window_time: row[window_time], true_sales: round(true_val, 2), pred_sales: round(max(pred_val, 0), 2), }) y_true.append(true_val) y_pred.append(pred_val) error compute_error_metrics(y_true, y_pred) spark.stop() return predictions, error except Exception as e: print(f[Spark ML] 预测失败: {e}) return None, None def save_predictions_to_db(predictions: list, error: dict): 保存预测结果和误差指标到数据库 from database import SessionLocal from models.prediction import Prediction from models.error_metric import ErrorMetric db SessionLocal() try: # 清空旧预测数据 db.query(Prediction).delete() for p in predictions: db.add(Prediction( window_timep[window_time], true_salesDecimal(str(p[true_sales])), pred_salesDecimal(str(p[pred_sales])), )) db.add(ErrorMetric( rmseDecimal(str(error[rmse])), maeDecimal(str(error[mae])), mapeDecimal(str(error[mape])), )) db.commit() print(f[Spark ML] 已保存 {len(predictions)} 条预测结果) finally: db.close()template div classpage-container div classpage-card div classpage-title销售额预测分析/div !-- 误差指标卡片 -- div classerror-cards div classerror-card div classmetric-labelRMSE (均方根误差)/div div classmetric-value{{ errorMetric.rmse }}/div /div div classerror-card div classmetric-labelMAE (平均绝对误差)/div div classmetric-value{{ errorMetric.mae }}/div /div div classerror-card div classmetric-labelMAPE (平均绝对百分比误差 %)/div div classmetric-value{{ errorMetric.mape }}%/div /div /div !-- 真实 vs 预测对比图 -- div refcompareRef classpred-chart pred-chart-compare/div !-- 残差图 -- div refresidualRef classpred-chart pred-chart-residual/div !-- 预测数据表格 -- el-table :datatableData stripe border stylewidth:100% el-table-column propwindow_time label时间窗口 min-width170 template #default{ row }{{ formatWindowTime(row.window_time) }}/template /el-table-column el-table-column proptrue_sales label真实销售额 min-width130 template #default{ row } span stylecolor:#409eff;font-weight:600¥{{ row.true_sales }}/span /template /el-table-column el-table-column proppred_sales label预测销售额 min-width130 template #default{ row } span stylecolor:#67c23a;font-weight:600¥{{ row.pred_sales }}/span /template /el-table-column el-table-column label误差 min-width120 template #default{ row } span :style{ color: Math.abs(row.true_sales - row.pred_sales) 500 ? #f56c6c : #909399 } ¥{{ (row.true_sales - row.pred_sales).toFixed(2) }} /span /template /el-table-column el-table-column propcreate_time label生成时间 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 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 [] /** * X 轴日期时间标签配置分行显示避免底部裁切 */ function buildAxisLabel() { return { rotate: 30, interval: auto, hideOverlap: true, fontSize: 11, margin: 16, formatter(val) { const text formatWindowTime(val) if (text.length 16) return ${text.slice(0, 10)}\n${text.slice(11)} return text }, } } /** * 初始化真实销售额 vs 预测销售额对比图 */ function initCompareChart(data) { const chart echarts.init(compareRef.value) const labels data.map(d formatWindowTime(d.window_time)) chart.setOption({ title: { text: 真实销售额 vs 预测销售额 对比, left: center, textStyle: { fontSize: 15 } }, tooltip: { trigger: axis, formatter(params) { const idx params[0]?.dataIndex ?? 0 const lines [labels[idx] || ] params.forEach(p lines.push(${p.marker}${p.seriesName}: ${p.value})) return lines.join(br/) }, }, // 图例放顶部避免与底部日期重叠 legend: { data: [真实销售额, 预测销售额], top: 32 }, xAxis: { type: category, data: labels, axisTick: { alignWithLabel: true }, axisLabel: buildAxisLabel(), }, yAxis: { type: value, name: 销售额(元) }, series: [ { name: 真实销售额, type: line, smooth: true, data: data.map(d Number(d.true_sales)), itemStyle: { color: #409eff }, lineStyle: { width: 3 }, symbol: circle, symbolSize: 8, }, { name: 预测销售额, type: line, smooth: true, data: data.map(d Number(d.pred_sales)), itemStyle: { color: #67c23a }, lineStyle: { width: 3, type: dashed }, symbol: diamond, symbolSize: 8, }, ], grid: { left: 20, right: 24, bottom: 28, top: 72, containLabel: true }, }) charts.push(chart) } /** * 初始化预测残差分析图 */ function initResidualChart(data) { const chart echarts.init(residualRef.value) const labels data.map(d formatWindowTime(d.window_time)) chart.setOption({ title: { text: 预测残差分析 (真实值 - 预测值), left: center, textStyle: { fontSize: 15 } }, tooltip: { trigger: axis, formatter(params) { const idx params[0]?.dataIndex ?? 0 const p params[0] return ${labels[idx] || }br/${p.marker}残差: ${p.value} }, }, xAxis: { type: category, data: labels, axisTick: { alignWithLabel: true }, axisLabel: buildAxisLabel(), }, yAxis: { type: value, name: 残差(元) }, series: [{ type: bar, data: data.map(d ({ value: d.residual, itemStyle: { color: d.residual 0 ? #409eff : #f56c6c }, })), barWidth: 20, }], grid: { left: 20, right: 24, bottom: 28, top: 56, containLabel: true }, }) charts.push(chart) } /** * 加载预测图表与误差指标 */ async function loadData() { const [errorRes, compareRes, residualRes] await Promise.all([ request.get(/prediction/error), request.get(/prediction/compare), request.get(/prediction/residual), ]) errorMetric.value errorRes.data charts.forEach(c c.dispose()) charts [] initCompareChart(compareRes.data) initResidualChart(residualRes.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() }) onUnmounted(() charts.forEach(c c.dispose())) /script style scoped /* 预留足够高度保证倾斜日期时间不被裁切 */ .pred-chart { width: 100%; margin-bottom: 24px; } .pred-chart-compare { height: 480px; } .pred-chart-residual { height: 420px; } /style
大家好我是Java1234_小锋老师分享一套锋哥原创的基于Spark实时电商用户行为分析与预测 系统(Python版本pyspark可视化大屏KafkaFastAPIVue3)项目介绍随着电子商务规模持续扩大用户在浏览、加购、收藏与购买等环节产生的行为数据呈现高并发、高吞吐与强时效特征。传统离线批处理分析难以满足运营决策对实时性的要求。本文设计并实现了一套基于 Spark 的实时电商用户行为分析与预测系统围绕“数据采集—流式计算—指标落库—可视化展示—销售预测”的完整链路展开研究与工程实践。系统采用前后端分离架构前端基于 Vue3、Element Plus 与 ECharts 构建管理端与数据大屏后端采用 Python FastAPI 提供 RESTful 接口并结合 JWT 完成管理员身份认证实时链路以 Kafka 作为消息中间件承接行为事件以 Spark Structured Streaming 完成按小时窗口的 PV、UV、加购、收藏、购买与销售额聚合预测模块基于 Spark ML 线性回归对销售额序列进行建模并输出 RMSE、MAE、MAPE 等误差指标。数据持久化采用 MySQL数据库名为 db_ecommerce核心业务表均以 t_ 前缀命名。测试结果表明系统能够稳定完成管理员登录、个人中心维护、行为与商品管理、实时统计展示、销售预测对比及流水线状态监控等功能具备较好的可扩展性与教学示范价值可为电商运营提供实时洞察与辅助决策支持。本文的主要工作包括完成系统需求分析与总体架构设计绘制实体属性图与实体关系图并完成八张核心业务表设计实现基于 Kafka 与 Spark 的实时统计及销售预测链路完成 Vue3 管理端与数据大屏开展功能测试并给出改进方向。研究结果表明将流式计算与 Web 管理系统结合能够在本科毕业设计条件下形成完整、可运行、可解释的实时分析应用。源码下载链接: https://pan.baidu.com/s/1u0yzt7SCx13nEjaH4FSMLw?pwd1234提取码: 1234系统展示核心代码 Spark ML 销售额预测模块 import numpy as np from decimal import Decimal from config import settings def compute_error_metrics(y_true: list, y_pred: list) - dict: 计算误差指标RMSE、MAE、MAPE y_true np.array(y_true, dtypefloat) y_pred np.array(y_pred, dtypefloat) rmse float(np.sqrt(np.mean((y_true - y_pred) ** 2))) mae float(np.mean(np.abs(y_true - y_pred))) mask y_true ! 0 if mask.any(): mape float(np.mean(np.abs((y_true[mask] - y_pred[mask]) / y_true[mask])) * 100) else: mape 0.0 return {rmse: round(rmse, 4), mae: round(mae, 4), mape: round(mape, 4)} def run_spark_prediction(sales_series: list None) - tuple: 使用 Spark ML 进行销售额预测 返回 (predictions, error_metrics) try: from pyspark.sql import SparkSession from pyspark.ml.feature import VectorAssembler from pyspark.ml.regression import LinearRegression from pyspark.sql.types import StructType, StructField, DoubleType, IntegerType, StringType import pyspark.sql.functions as F spark SparkSession.builder \ .appName(SalesPrediction) \ .master(settings.SPARK_MASTER) \ .config(spark.driver.memory, 2g) \ .getOrCreate() spark.sparkContext.setLogLevel(WARN) if sales_series is None: from database import SessionLocal from models.realtime_stat import RealtimeStat db SessionLocal() stats db.query(RealtimeStat).order_by(RealtimeStat.window_time.asc()).all() db.close() sales_series [ {window_time: s.window_time, sales: float(s.sales)} for s in stats ] if len(sales_series) 5: spark.stop() return [], {rmse: 0, mae: 0, mape: 0} # 构造滞后特征 data [] for i in range(3, len(sales_series)): data.append({ window_time: sales_series[i][window_time], lag1: sales_series[i - 1][sales], lag2: sales_series[i - 2][sales], lag3: sales_series[i - 3][sales], hour: int(str(sales_series[i][window_time])[11:13]), sales: sales_series[i][sales], }) schema StructType([ StructField(window_time, StringType()), StructField(lag1, DoubleType()), StructField(lag2, DoubleType()), StructField(lag3, DoubleType()), StructField(hour, IntegerType()), StructField(sales, DoubleType()), ]) df spark.createDataFrame(data, schema) # 划分训练集和测试集后20%作为测试 split_idx max(int(len(data) * 0.8), 1) train_df df.limit(split_idx) test_df df.filter(F.monotonically_increasing_id() split_idx) assembler VectorAssembler( inputCols[lag1, lag2, lag3, hour], outputColfeatures ) train_df assembler.transform(train_df) test_df assembler.transform(test_df) lr LinearRegression(featuresColfeatures, labelColsales, maxIter100) model lr.fit(train_df) predictions_raw model.transform(test_df).collect() predictions [] y_true, y_pred [], [] for row in predictions_raw: true_val float(row[sales]) pred_val float(row[prediction]) predictions.append({ window_time: row[window_time], true_sales: round(true_val, 2), pred_sales: round(max(pred_val, 0), 2), }) y_true.append(true_val) y_pred.append(pred_val) error compute_error_metrics(y_true, y_pred) spark.stop() return predictions, error except Exception as e: print(f[Spark ML] 预测失败: {e}) return None, None def save_predictions_to_db(predictions: list, error: dict): 保存预测结果和误差指标到数据库 from database import SessionLocal from models.prediction import Prediction from models.error_metric import ErrorMetric db SessionLocal() try: # 清空旧预测数据 db.query(Prediction).delete() for p in predictions: db.add(Prediction( window_timep[window_time], true_salesDecimal(str(p[true_sales])), pred_salesDecimal(str(p[pred_sales])), )) db.add(ErrorMetric( rmseDecimal(str(error[rmse])), maeDecimal(str(error[mae])), mapeDecimal(str(error[mape])), )) db.commit() print(f[Spark ML] 已保存 {len(predictions)} 条预测结果) finally: db.close()template div classpage-container div classpage-card div classpage-title销售额预测分析/div !-- 误差指标卡片 -- div classerror-cards div classerror-card div classmetric-labelRMSE (均方根误差)/div div classmetric-value{{ errorMetric.rmse }}/div /div div classerror-card div classmetric-labelMAE (平均绝对误差)/div div classmetric-value{{ errorMetric.mae }}/div /div div classerror-card div classmetric-labelMAPE (平均绝对百分比误差 %)/div div classmetric-value{{ errorMetric.mape }}%/div /div /div !-- 真实 vs 预测对比图 -- div refcompareRef classpred-chart pred-chart-compare/div !-- 残差图 -- div refresidualRef classpred-chart pred-chart-residual/div !-- 预测数据表格 -- el-table :datatableData stripe border stylewidth:100% el-table-column propwindow_time label时间窗口 min-width170 template #default{ row }{{ formatWindowTime(row.window_time) }}/template /el-table-column el-table-column proptrue_sales label真实销售额 min-width130 template #default{ row } span stylecolor:#409eff;font-weight:600¥{{ row.true_sales }}/span /template /el-table-column el-table-column proppred_sales label预测销售额 min-width130 template #default{ row } span stylecolor:#67c23a;font-weight:600¥{{ row.pred_sales }}/span /template /el-table-column el-table-column label误差 min-width120 template #default{ row } span :style{ color: Math.abs(row.true_sales - row.pred_sales) 500 ? #f56c6c : #909399 } ¥{{ (row.true_sales - row.pred_sales).toFixed(2) }} /span /template /el-table-column el-table-column propcreate_time label生成时间 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 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 [] /** * X 轴日期时间标签配置分行显示避免底部裁切 */ function buildAxisLabel() { return { rotate: 30, interval: auto, hideOverlap: true, fontSize: 11, margin: 16, formatter(val) { const text formatWindowTime(val) if (text.length 16) return ${text.slice(0, 10)}\n${text.slice(11)} return text }, } } /** * 初始化真实销售额 vs 预测销售额对比图 */ function initCompareChart(data) { const chart echarts.init(compareRef.value) const labels data.map(d formatWindowTime(d.window_time)) chart.setOption({ title: { text: 真实销售额 vs 预测销售额 对比, left: center, textStyle: { fontSize: 15 } }, tooltip: { trigger: axis, formatter(params) { const idx params[0]?.dataIndex ?? 0 const lines [labels[idx] || ] params.forEach(p lines.push(${p.marker}${p.seriesName}: ${p.value})) return lines.join(br/) }, }, // 图例放顶部避免与底部日期重叠 legend: { data: [真实销售额, 预测销售额], top: 32 }, xAxis: { type: category, data: labels, axisTick: { alignWithLabel: true }, axisLabel: buildAxisLabel(), }, yAxis: { type: value, name: 销售额(元) }, series: [ { name: 真实销售额, type: line, smooth: true, data: data.map(d Number(d.true_sales)), itemStyle: { color: #409eff }, lineStyle: { width: 3 }, symbol: circle, symbolSize: 8, }, { name: 预测销售额, type: line, smooth: true, data: data.map(d Number(d.pred_sales)), itemStyle: { color: #67c23a }, lineStyle: { width: 3, type: dashed }, symbol: diamond, symbolSize: 8, }, ], grid: { left: 20, right: 24, bottom: 28, top: 72, containLabel: true }, }) charts.push(chart) } /** * 初始化预测残差分析图 */ function initResidualChart(data) { const chart echarts.init(residualRef.value) const labels data.map(d formatWindowTime(d.window_time)) chart.setOption({ title: { text: 预测残差分析 (真实值 - 预测值), left: center, textStyle: { fontSize: 15 } }, tooltip: { trigger: axis, formatter(params) { const idx params[0]?.dataIndex ?? 0 const p params[0] return ${labels[idx] || }br/${p.marker}残差: ${p.value} }, }, xAxis: { type: category, data: labels, axisTick: { alignWithLabel: true }, axisLabel: buildAxisLabel(), }, yAxis: { type: value, name: 残差(元) }, series: [{ type: bar, data: data.map(d ({ value: d.residual, itemStyle: { color: d.residual 0 ? #409eff : #f56c6c }, })), barWidth: 20, }], grid: { left: 20, right: 24, bottom: 28, top: 56, containLabel: true }, }) charts.push(chart) } /** * 加载预测图表与误差指标 */ async function loadData() { const [errorRes, compareRes, residualRes] await Promise.all([ request.get(/prediction/error), request.get(/prediction/compare), request.get(/prediction/residual), ]) errorMetric.value errorRes.data charts.forEach(c c.dispose()) charts [] initCompareChart(compareRes.data) initResidualChart(residualRes.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() }) onUnmounted(() charts.forEach(c c.dispose())) /script style scoped /* 预留足够高度保证倾斜日期时间不被裁切 */ .pred-chart { width: 100%; margin-bottom: 24px; } .pred-chart-compare { height: 480px; } .pred-chart-residual { height: 420px; } /style
