1. 选题背景与整体设计思路1.1 为什么“金融新闻情感 DAX40 Spark”是个好选题先说说我为什么觉得这个题目值得做。金融领域有一个被反复验证的现象股价的短期波动很大程度上受市场情绪驱动而不是单纯由财报数据决定。一条关于某家公司的负面新闻可能在几分钟内就让它的股价跌掉两三个百分点。DAX40是德国法兰克福证券交易所的蓝筹股指数涵盖40家德国头部上市公司像西门子、宝马、拜耳、SAP这些名字大家都不陌生。这些公司的新闻曝光度高、数据可得性好非常适合做情感分析类课题。那为什么非要上Spark很多同学第一反应是“我用pandas也能跑啊”。没错几千条新闻用pandas确实够用。但问题在于金融新闻的数据量级和更新频率远超想象。你如果要做一年的滚动分析每天抓取DAX40全部成分股的相关新闻轻松就是几十万到上百万条文本。再加上你要做分词、情感打分、时间窗口聚合、和股价数据做关联pandas单机跑起来会非常吃力。Spark的分布式计算能力在这里就有了用武之地而且“大数据框架 金融场景”的组合在毕设答辩时天然比“爬虫 pandas画图”更有说服力。这个选题适合什么人做我总结下来是三类第一类数据科学或计算机专业想做一个有真实业务背景的毕设第二类金融工程或量化方向想补足自己在文本挖掘上的技术短板第三类已经工作但想转行做数据分析需要一个能写进简历的完整项目。不管你是哪一类这个系统的核心链路都是通用的数据采集 → 数据清洗 → 情感分析 → 趋势聚合 → 可视化呈现。1.2 系统整体架构怎么搭我在设计这类系统时习惯先把数据流画清楚再决定每一层用什么技术。整个系统的架构可以分成五层我用一个表格把每层的职责和选型理由说清楚。层级职责技术选型选型理由数据采集层抓取DAX40成分股相关新闻Python requests BeautifulSoup / NewsAPI轻量、可控适合毕设规模数据存储层原始数据与中间结果存储HDFS Hive / 本地ParquetSpark原生支持Parquet读写快计算分析层分词、情感打分、聚合Spark SQL Spark MLlib分布式处理支持大规模文本情感模型层情感分类预训练模型 传统机器学习对比兼顾效果与可解释性可视化层趋势展示Flask ECharts / Streamlit快速搭建交互式看板这里我要特别说明一下存储层的选择。很多同学一上来就想着上Hive觉得“大数据项目必须配Hive”。但实际上如果你的数据量在百万级以内直接用Spark读写Parquet文件反而更简单、更快。Parquet是列式存储做聚合查询时只读需要的列I/O效率比CSV高一个数量级。我建议的做法是原始新闻存一份JSON清洗后的结构化数据存Parquet情感分析结果再存一份Parquet。这样每一层都可以独立复现调试起来也方便。计算分析层是整个系统的核心。Spark SQL负责结构化查询和聚合比如“按周统计每只股票的平均情感得分”Spark MLlib负责机器学习部分比如用TF-IDF 逻辑回归做一个基线情感分类器。如果你想让系统更有深度可以再加一个深度学习模型做对比但要注意毕设的时间成本不要贪多。1.3 情感分析方案怎么选这是整个系统里最容易踩坑的地方。我见过太多同学直接拿一个中文情感分析模型去跑英文金融新闻结果准确率惨不忍睹。DAX40的新闻主要是英文和德文所以你的模型必须支持这两种语言或者至少支持英文。我的建议是分两步走。第一步先用一个成熟的开源工具做基线比如TextBlob或者VADER。VADER是专门为社交媒体文本设计的情感分析工具对新闻标题这种短文本效果不错而且不需要训练开箱即用。第二步如果你想让系统更有技术含量可以自己标注一批数据训练一个逻辑回归或SVM分类器和VADER的结果做对比。这样你的毕设就有了“基线 vs 改进”的对比实验答辩时非常加分。如果你非要用深度学习我推荐用Hugging Face上的预训练模型比如distilbert-base-uncased-finetuned-sst-2-english它在英文情感分类上表现稳定而且模型小、推理快。但要注意金融领域的情感表达和通用领域差别很大。比如“the stock plummeted”在通用模型里可能只是“负面”但在金融语境下是“强烈负面”。所以如果你要做精细化的情感强度分析最好还是自己微调一下模型。2. 核心细节解析与实操要点2.1 DAX40成分股数据怎么获取和更新DAX40的成分股不是一成不变的。德国指数公司Qontigo每季度会评估一次可能调整成分股。所以你的系统必须有一个机制来维护这份列表。我通常的做法是在项目里放一个dax40_constituents.csv文件包含股票代码、公司名称、行业分类三个字段。这个文件不需要频繁更新每季度手动检查一次即可。获取新闻数据的渠道有几个选择。如果你在学校可以看看图书馆有没有购买新闻数据库的API权限比如Bloomberg或Refinitiv但这类接口通常有严格的访问限制。更实际的做法是用公开的新闻API比如NewsAPI、GNews或者Marketaux。Marketaux专门做金融新闻还自带情感标签但免费额度有限。我的建议是用NewsAPI做主力每天定时抓取用RSS订阅做补充比如路透社、彭博社的财经RSS。抓取的时候要注意几个细节。第一关键词要设计好。不要只搜“DAX40”要针对每只成分股单独搜比如“Siemens news”、“BMW stock”。第二时间范围要明确。毕设通常做一年到两年的数据就够了太长反而增加清洗难度。第三去重很重要。同一篇新闻可能被多个来源转载你需要用标题的SimHash或者URL的MD5来做去重。# 新闻去重示例基于标题的SimHash from simhash import Simhash def get_simhash(text): return Simhash(text.split()).value def deduplicate_news(news_list, threshold3): seen [] unique_news [] for news in news_list: h get_simhash(news[title]) if not any(abs(h - s) threshold for s in seen): seen.append(h) unique_news.append(news) return unique_news2.2 Spark集群怎么搭才不踩坑这是很多同学最头疼的部分。我见过太多人在“Spark集群搭建”这一步卡了一两周最后放弃改用单机模式。其实对于毕设来说你不需要一个真正的多节点集群。用Docker搭一个伪分布式环境或者直接在单机上用local[*]模式跑完全够用。关键是你要理解Spark的运行机制而不是真的去维护一个集群。如果你确实想搭一个多节点集群我建议用Docker Compose一台机器上起三个容器一个Master两个Worker。这样既能看到分布式的效果又不用折腾网络配置。内存分配上每个Worker给2GBMaster给1GB对于百万级数据完全够用。注意Spark的内存管理有两个概念spark.executor.memory和spark.driver.memory。前者是每个执行器的内存后者是驱动进程的内存。做聚合操作时如果数据量大要适当调大spark.sql.shuffle.partitions默认是200对于小数据集反而会拖慢速度可以调到50左右。# Docker Compose启动Spark集群的简化配置 version: 3 services: spark-master: image: bitnami/spark:3.5 environment: - SPARK_MODEmaster - SPARK_MASTER_HOSTspark-master ports: - 8080:8080 - 7077:7077 spark-worker-1: image: bitnami/spark:3.5 environment: - SPARK_MODEworker - SPARK_MASTER_URLspark://spark-master:7077 - SPARK_WORKER_MEMORY2G spark-worker-2: image: bitnami/spark:3.5 environment: - SPARK_MODEworker - SPARK_MASTER_URLspark://spark-master:7077 - SPARK_WORKER_MEMORY2G注意如果你用Windows系统Spark的本地模式需要额外配置Hadoop的winutils.exe否则会报java.io.IOException: Could not locate executable null\bin\winutils.exe。这个坑我踩过解决办法是下载对应Hadoop版本的winutils放到一个目录下然后设置HADOOP_HOME环境变量。2.3 文本预处理的关键步骤金融新闻的文本预处理和普通文本挖掘不太一样。普通文本挖掘可能只需要分词、去停用词但金融新闻有几个特殊之处。第一股票代码和公司简称需要统一。比如“BMW”、“Bayerische Motoren Werke”、“宝马”可能出现在同一篇新闻里你需要把它们映射到同一个实体。第二数字和百分比很重要。“revenue increased 15%”和“revenue decreased 15%”情感完全相反所以不能简单地把数字去掉。第三金融术语需要保留。“bullish”、“bearish”、“volatile”这些词是情感信号不能当停用词过滤掉。我的预处理流程是这样的先做HTML标签清洗和特殊字符替换然后做实体归一化接着分词和词性标注最后做停用词过滤但保留金融情感词。分词工具推荐用spaCy它对英文的处理比NLTK更稳定而且支持命名实体识别可以自动识别公司名和股票代码。import spacy nlp spacy.load(en_core_web_sm) def preprocess_news(text): # 清洗HTML标签 text re.sub(r[^], , text) # 实体归一化 text text.replace(Bayerische Motoren Werke, BMW) text text.replace(Volkswagen AG, VW) # spaCy处理 doc nlp(text) tokens [] for token in doc: if token.is_stop and token.text.lower() not in FINANCIAL_SENTIMENT_WORDS: continue if token.is_punct: continue tokens.append(token.lemma_.lower()) return .join(tokens)这里有个细节要注意spaCy的is_stop会过滤掉“not”、“no”这类否定词但在情感分析里否定词极其重要。“not good”和“good”情感完全相反。所以你需要自定义一个停用词表把否定词和金融情感词排除在外。我通常会把“not”、“no”、“never”、“neither”、“nor”这几个词加入白名单确保它们不被过滤。3. 实操过程与核心环节实现3.1 数据采集与存储的完整流程我以NewsAPI为例走一遍完整的数据采集流程。首先你需要注册一个账号拿到API Key。免费版每天有100次请求每次最多返回100条新闻。对于DAX40的40只股票你每天需要40次请求刚好够用。但要注意免费版只能查最近一个月的新闻所以如果你想做一年的数据需要每天定时抓取并累积。import requests import json from datetime import datetime, timedelta API_KEY your_api_key_here BASE_URL https://newsapi.org/v2/everything def fetch_news(query, from_date, to_date): params { q: query, from: from_date, to: to_date, language: en, sortBy: publishedAt, apiKey: API_KEY } response requests.get(BASE_URL, paramsparams) if response.status_code 200: return response.json().get(articles, []) else: print(fError: {response.status_code}) return [] # 按股票逐只抓取 stocks [Siemens, BMW, SAP, Bayer, Volkswagen] all_news [] for stock in stocks: news fetch_news(stock, 2024-01-01, 2024-12-31) for item in news: item[stock] stock all_news.extend(news) print(f{stock}: {len(news)} articles)抓下来的数据先存成JSON每天一个文件命名格式用news_YYYYMMDD.json。这样做的好处是如果某天抓取失败你可以单独重跑那一天的不用全部重来。存到HDFS或者本地文件系统都可以Spark都能读。接下来是数据清洗。原始新闻里有很多噪声比如广告、重复内容、无关股票。我通常用Spark做这一步因为清洗规则可能很复杂用Spark SQL写起来更清晰。from pyspark.sql import SparkSession from pyspark.sql.functions import col, lower, regexp_replace, length spark SparkSession.builder \ .appName(DAX40NewsCleaning) \ .config(spark.sql.shuffle.partitions, 50) \ .getOrCreate() # 读取原始JSON raw_df spark.read.json(hdfs:///data/news/*.json) # 清洗去重、去短文本、去广告 cleaned_df raw_df \ .dropDuplicates([title, publishedAt]) \ .filter(length(col(content)) 100) \ .filter(~lower(col(title)).contains(advertisement)) \ .withColumn(content, regexp_replace(col(content), r[^], )) cleaned_df.write.mode(overwrite).parquet(hdfs:///data/news_cleaned/)这里有个经验dropDuplicates在Spark里是基于全列比较的如果只想按标题去重需要先select出标题列去重后再join回去。或者用dropDuplicates([title])指定列但要注意这会保留第一条出现的记录不保证是最新的。如果你需要保留最新的可以先按时间排序再去重。3.2 情感分析模型的训练与推理我建议做两个模型一个基线模型一个改进模型。基线用VADER不需要训练直接调用。改进模型用Spark MLlib的逻辑回归需要自己标注数据。标注数据是个体力活。我的做法是从清洗后的新闻里随机抽1000条人工标注为正面、负面、中性三类。标注标准要提前定好比如“股价上涨”是正面“股价下跌”是负面“公司发布财报”是中性。标注完成后用TF-IDF提取特征训练逻辑回归。from pyspark.ml.feature import Tokenizer, HashingTF, IDF from pyspark.ml.classification import LogisticRegression from pyspark.ml import Pipeline # 假设labeled_df有content和label两列 tokenizer Tokenizer(inputColcontent, outputColwords) hashingTF HashingTF(inputColwords, outputColrawFeatures, numFeatures10000) idf IDF(inputColrawFeatures, outputColfeatures) lr LogisticRegression(maxIter20, regParam0.01) pipeline Pipeline(stages[tokenizer, hashingTF, idf, lr]) model pipeline.fit(train_df) # 推理 predictions model.transform(test_df) predictions.select(content, label, prediction).show(10)VADER的用法更简单但它只能处理单条文本不能直接用在Spark DataFrame上。你需要用UDF用户自定义函数包装一下。from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer from pyspark.sql.functions import udf from pyspark.sql.types import FloatType analyzer SentimentIntensityAnalyzer() def get_sentiment(text): if text is None: return 0.0 return analyzer.polarity_scores(text)[compound] sentiment_udf udf(get_sentiment, FloatType()) df_with_sentiment cleaned_df.withColumn(sentiment, sentiment_udf(col(content)))注意VADER的compound分数范围是-1到1-1表示极度负面1表示极度正面。但金融新闻里很多中性报道的分数可能在-0.1到0.1之间你需要根据实际分布调整阈值。我通常把大于0.05的算正面小于-0.05的算负面中间算中性。3.3 趋势聚合与可视化呈现情感分析做完之后你需要把结果按时间维度聚合。最常见的做法是按周或按月统计每只股票的平均情感得分然后和股价走势做对比。这里的关键是时间对齐。新闻的发布时间和股价的交易时间可能不一致比如周末的新闻会影响周一的股价。我的处理方式是把新闻按自然周聚合股价也按自然周取平均值这样对齐更合理。from pyspark.sql.functions import weekofyear, year, avg, count # 按周聚合情感得分 weekly_sentiment df_with_sentiment \ .withColumn(year, year(col(publishedAt))) \ .withColumn(week, weekofyear(col(publishedAt))) \ .groupBy(stock, year, week) \ .agg(avg(sentiment).alias(avg_sentiment), count(*).alias(news_count)) weekly_sentiment.write.mode(overwrite).parquet(hdfs:///data/weekly_sentiment/)可视化部分我推荐用Streamlit因为它上手快几行代码就能出一个交互式看板。你可以做一个折线图横轴是时间纵轴是情感得分再叠加股价走势。如果两条线有明显的相关性你的毕设就有了一个很有说服力的结论。import streamlit as st import pandas as pd import plotly.express as px st.title(DAX40金融新闻情感趋势分析) # 读取聚合结果 df pd.read_parquet(weekly_sentiment.parquet) stock st.selectbox(选择股票, df[stock].unique()) filtered df[df[stock] stock] fig px.line(filtered, xweek, yavg_sentiment, titlef{stock} 周度情感趋势) st.plotly_chart(fig)如果你想让可视化更高级可以做一个热力图横轴是时间纵轴是股票颜色表示情感得分。这样一眼就能看出哪些股票在哪些时间段情绪异常。ECharts的热力图配置稍微复杂一点但效果很好网上有很多模板可以直接改。4. 常见问题与排查技巧实录4.1 Spark运行中的典型报错与解决做这个项目你一定会遇到各种Spark报错。我整理了一个速查表覆盖了最常见的几个问题。报错信息原因解决方法java.lang.OutOfMemoryError: Java heap space执行器内存不足调大spark.executor.memory或减少spark.sql.shuffle.partitionsorg.apache.spark.SparkException: Task not serializableUDF里引用了不可序列化的对象把对象定义在UDF内部或用broadcast广播java.io.IOException: Could not locate executable null\bin\winutils.exeWindows下缺少Hadoop工具下载winutils设置HADOOP_HOMEAnalysisException: Path does not exist文件路径错误检查HDFS路径或本地路径是否正确Py4JJavaError: An error occurred while calling oXX.showString通常是数据倾斜对key加盐或改用reduceByKey替代groupByKey数据倾斜是Spark里最隐蔽的问题。比如你按股票分组但某只股票的新闻量特别大其他股票很小那么处理这只股票的任务就会特别慢。解决办法是给key加一个随机前缀先做局部聚合再去掉前缀做全局聚合。这个技巧在Spark SQL里可以用salting实现但写起来比较麻烦。如果数据量不是特别大我建议直接用repartition重新分区让数据分布更均匀。4.2 情感分析准确率上不去的排查思路如果你发现模型准确率只有60%左右别急着换模型先检查数据。我见过最常见的问题是标注标准不一致。比如同一条新闻你标了正面你的同学标了中性。这种噪声数据会让模型学不到规律。解决办法是先制定一个详细的标注指南找两个人独立标注100条计算一下一致性。如果一致性低于80%说明标注标准需要细化。第二个常见问题是类别不平衡。金融新闻里中性报道占大多数正面和负面各占一小部分。如果直接用原始数据训练模型会倾向于预测中性。解决办法是做重采样或者用类别权重。Spark MLlib的逻辑回归支持weightCol参数你可以给少数类更高的权重。第三个问题是特征工程不到位。TF-IDF只考虑了词频没有考虑词序和上下文。如果你想让效果更好可以加入n-gram特征或者用预训练的词向量。但要注意毕设的时间有限不要在一个点上钻太深。我的建议是先把基线跑通再逐步优化每一步都要有对比实验。4.3 数据采集的合规与稳定性问题新闻API通常有频率限制免费版尤其严格。如果你不小心超了可能会被封IP。我的做法是在代码里加一个请求间隔比如每次请求后sleep(2)秒。另外要做好错误重试。网络请求失败是常态不要因为一次失败就放弃。用try-except包起来失败后等几秒重试重试三次还失败就记录日志跳过这条。import time import logging logging.basicConfig(filenamefetch_errors.log, levellogging.ERROR) def fetch_with_retry(url, params, max_retries3): for i in range(max_retries): try: response requests.get(url, paramsparams, timeout10) if response.status_code 200: return response.json() elif response.status_code 429: time.sleep(60) # 频率限制等一分钟 else: time.sleep(5) except Exception as e: logging.error(fAttempt {i1} failed: {str(e)}) time.sleep(5) return None还有一个容易被忽略的问题是数据存储的持久化。如果你把数据存在本地电脑一关机就没了。我建议用云盘同步或者存到学校的服务器上。如果条件允许用HDFS是最稳妥的但配置起来麻烦。折中方案是用SQLite存原始数据Parquet存中间结果这样既轻量又可靠。4.4 毕设答辩时怎么讲清楚技术亮点最后说一个非技术但很重要的问题答辩怎么讲。我见过很多同学代码写得不错但一上台就只会念PPT把“我用了Spark”重复十遍。评委想听的不是你用了什么工具而是你为什么用这个工具解决了什么问题。我的建议是准备三个故事。第一个故事讲数据规模我抓了多少条新闻如果用pandas要跑多久用Spark只用了多久。第二个故事讲技术对比我试了VADER和逻辑回归准确率分别是多少为什么逻辑回归更好。第三个故事讲业务发现我发现某只股票的情感得分和股价走势有滞后相关性滞后大概几天。这三个故事讲完你的毕设就有了灵魂而不只是一个“跑通的系统”。另外答辩时一定要准备一个现场演示。不要只放截图要真的打开系统点几下让评委看到交互效果。如果担心现场网络问题提前录一个演示视频作为备份。这个细节很多人忽略但关键时刻能救你一命。提示如果你的毕设需要提交代码记得写一个清晰的README说明环境依赖、数据来源、运行步骤。评委可能不会跑你的代码但README的质量直接影响他对你工程能力的判断。我通常会在README里放一个“快速开始”章节三行命令就能跑起来这样即使是不懂技术的人也能感受到你的项目是完整的。