大家好，我是Java1234_小锋老师，分享一套锋哥原创的基于Spark实时社交媒体舆情分析与趋势预测(Python版本+pyspark+可视化大屏+Kafka+FastAPI+Vue3) 

项目介绍

随着微博、抖音、知乎、小红书等社交媒体的快速发展，网络舆情呈现出数据规模大、传播速度快、情感变化剧烈等特点。传统基于离线批处理的舆情分析方法难以满足“秒级感知、分钟级研判”的业务需求。针对上述问题，本文设计并实现了一套基于 Spark 的实时社交媒体舆情分析与趋势预测系统。系统采用前后端分离架构：前端基于 Vue3、Element Plus 与 ECharts 构建管理后台和可视化大屏；后端基于 Python 与 FastAPI 提供统一 REST API；消息层引入 Kafka 承接高并发舆情事件流；计算层使用 Spark Streaming（Structured Streaming）完成按小时窗口的帖文量、独立用户数、正中负情感分布与热度指数聚合；预测层基于 Spark ML 线性回归，结合滞后热度特征与小时特征，对舆情热度进行趋势预测，并以 RMSE、MAE、MAPE 评估模型误差。数据持久化采用 MySQL，数据库名为 db_social_opinion。系统还设计了 Kafka/Spark 不可用时的 pandas 与 scikit-learn 降级方案，保证演示与实验环境的可用性。测试结果表明，系统能够稳定完成舆情数据采集、实时统计、趋势预测与可视化展示，功能完整、结构清晰，达到本科毕业设计要求。

源码下载 

链接: https://pan.baidu.com/s/15-uuTt3lRlIFc0AzRH8ANw?pwd=1234
提取码: 1234

系统展示

核心代码

"""
Spark Streaming 流式计算 - 消费 Kafka 数据并实时统计舆情
"""
from config import settings


def run_spark_streaming(events: list = None) -> list:
    """
    运行 Spark Structured Streaming 处理社交媒体数据流
    若传入 events 列表则直接处理（用于降级模式复用逻辑）
    返回窗口统计结果列表
    """
    try:
        from pyspark.sql import SparkSession
        from pyspark.sql.functions import (
            col, count, sum as spark_sum, countDistinct,
            window, from_json, to_timestamp
        )
        from pyspark.sql.types import (
            StructType, StructField, StringType, IntegerType, DoubleType
        )

        spark = SparkSession.builder \
            .appName(settings.SPARK_APP_NAME) \
            .master(settings.SPARK_MASTER) \
            .config("spark.sql.shuffle.partitions", "4") \
            .config("spark.driver.memory", "2g") \
            .getOrCreate()
        spark.sparkContext.setLogLevel("WARN")

        schema = StructType([
            StructField("user_id", IntegerType()),
            StructField("platform_id", IntegerType()),
            StructField("topic_id", IntegerType()),
            StructField("content", StringType()),
            StructField("sentiment", StringType()),
            StructField("heat", DoubleType()),
            StructField("event_time", StringType()),
        ])

        if events:
            df = spark.createDataFrame(events)
        else:
            raw_df = spark.readStream \
                .format("kafka") \
                .option("kafka.bootstrap.servers", settings.KAFKA_BOOTSTRAP_SERVERS) \
                .option("subscribe", settings.KAFKA_TOPIC) \
                .option("startingOffsets", "earliest") \
                .load()

            df = raw_df.select(
                from_json(col("value").cast("string"), schema).alias("data")
            ).select("data.*")

        df = df.withColumn("ts", to_timestamp(col("event_time"), "yyyy-MM-dd HH:mm:ss"))
        df = df.filter(col("ts").isNotNull())

        windowed = df.groupBy(window(col("ts"), "1 hour")).agg(
            count("*").alias("post_count"),
            countDistinct("user_id").alias("uv"),
            spark_sum((col("sentiment") == "positive").cast("int")).alias("positive"),
            spark_sum((col("sentiment") == "neutral").cast("int")).alias("neutral"),
            spark_sum((col("sentiment") == "negative").cast("int")).alias("negative"),
            spark_sum("heat").alias("heat_index"),
        )

        if events:
            rows = windowed.collect()
            results = []
            for row in rows:
                start = row["window"].start
                results.append({
                    "window_time": start.strftime("%Y-%m-%d %H:00:00"),
                    "post_count": int(row["post_count"]),
                    "uv": int(row["uv"]),
                    "positive": int(row["positive"]),
                    "neutral": int(row["neutral"]),
                    "negative": int(row["negative"]),
                    "heat_index": float(row["heat_index"]),
                })
            spark.stop()
            return results

        spark.stop()
        return []

    except Exception as e:
        print(f"[Spark Streaming] 运行失败: {e}")
        return None


def save_stats_to_db(stats: list):
    """
    将统计结果写入数据库
    """
    from database import SessionLocal
    from models.realtime_stat import RealtimeStat

    db = SessionLocal()
    try:
        for s in stats:
            existing = db.query(RealtimeStat).filter(
                RealtimeStat.window_time == s["window_time"]
            ).first()
            if existing:
                for k, v in s.items():
                    setattr(existing, k, v)
            else:
                db.add(RealtimeStat(**s))
        db.commit()
        print(f"[Spark Streaming] 已写入 {len(stats)} 条统计结果")
    finally:
        db.close()

<template>
  <div class="page-container">
    <div class="page-card">
      <div class="page-title">舆情热度预测分析</div>
      <div class="error-cards">
        <div class="error-card">
          <div class="metric-label">RMSE (均方根误差)</div>
          <div class="metric-value">{{ errorMetric.rmse }}</div>
        </div>
        <div class="error-card">
          <div class="metric-label">MAE (平均绝对误差)</div>
          <div class="metric-value">{{ errorMetric.mae }}</div>
        </div>
        <div class="error-card">
          <div class="metric-label">MAPE (平均绝对百分比误差 %)</div>
          <div class="metric-value">{{ errorMetric.mape }}%</div>
        </div>
      </div>
      <div ref="compareRef" class="pred-chart pred-chart-compare"></div>
      <div ref="residualRef" class="pred-chart pred-chart-residual"></div>
      <el-table :data="tableData" stripe border style="width:100%">
        <el-table-column prop="window_time" label="时间窗口" min-width="170">
          <template #default="{ row }">{{ formatWindowTime(row.window_time) }}</template>
        </el-table-column>
        <el-table-column prop="true_heat" label="真实热度" min-width="120">
          <template #default="{ row }"><span style="color:#409eff;font-weight:600">{{ row.true_heat }}</span></template>
        </el-table-column>
        <el-table-column prop="pred_heat" label="预测热度" min-width="120">
          <template #default="{ row }"><span style="color:#67c23a;font-weight:600">{{ row.pred_heat }}</span></template>
        </el-table-column>
        <el-table-column label="误差" min-width="100">
          <template #default="{ row }">
            <span :style="{ color: Math.abs(row.true_heat - row.pred_heat) > 50 ? '#f56c6c' : '#909399' }">
              {{ (row.true_heat - row.pred_heat).toFixed(2) }}
            </span>
          </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>
      <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="loadTable" />
    </div>
  </div>
</template>

<script setup>
/**
 * 预测分析页面：真实 vs 预测对比图 + 误差分析
 */
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 = []

function buildAxisLabel() {
  return {
    rotate: 30, interval: 'auto', fontSize: 11, margin: 16,
    formatter(val) { const text = formatWindowTime(val); return text.length >= 16 ? `${text.slice(0,10)}\n${text.slice(11)}` : text },
  }
}

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' },
    legend: { data: ['真实热度', '预测热度'], top: 32 },
    xAxis: { type: 'category', data: labels, axisLabel: buildAxisLabel() },
    yAxis: { type: 'value', name: '热度指数' },
    series: [
      { name: '真实热度', type: 'line', smooth: true, data: data.map(d => Number(d.true_heat)), itemStyle: { color: '#409eff' }, lineStyle: { width: 3 } },
      { name: '预测热度', type: 'line', smooth: true, data: data.map(d => Number(d.pred_heat)), itemStyle: { color: '#67c23a' }, lineStyle: { width: 3, type: 'dashed' } },
    ],
    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' },
    xAxis: { type: 'category', data: labels, 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>