1. 选题背景与整体设计思路

1.1 为什么“金融新闻情感 + DAX40 + Spark”是个好选题

先说说我为什么觉得这个题目值得做。金融领域有一个被反复验证的现象：股价的短期波动，很大程度上受市场情绪驱动，而不是单纯由财报数据决定。一条关于某家公司的负面新闻，可能在几分钟内就让它的股价跌掉两三个百分点。DAX40是德国法兰克福证券交易所的蓝筹股指数，涵盖40家德国头部上市公司，像西门子、宝马、拜耳、SAP这些名字大家都不陌生。这些公司的新闻曝光度高、数据可得性好，非常适合做情感分析类课题。

那为什么非要上Spark？很多同学第一反应是“我用pandas也能跑啊”。没错，几千条新闻用pandas确实够用。但问题在于，金融新闻的数据量级和更新频率远超想象。你如果要做一年的滚动分析，每天抓取DAX40全部成分股的相关新闻，轻松就是几十万到上百万条文本。再加上你要做分词、情感打分、时间窗口聚合、和股价数据做关联，pandas单机跑起来会非常吃力。Spark的分布式计算能力在这里就有了用武之地，而且“大数据框架 + 金融场景”的组合，在毕设答辩时天然比“爬虫 + pandas画图”更有说服力。

这个选题适合什么人做？我总结下来是三类：第一类，数据科学或计算机专业，想做一个有真实业务背景的毕设；第二类，金融工程或量化方向，想补足自己在文本挖掘上的技术短板；第三类，已经工作但想转行做数据分析，需要一个能写进简历的完整项目。不管你是哪一类，这个系统的核心链路都是通用的：数据采集 → 数据清洗 → 情感分析 → 趋势聚合 → 可视化呈现。

1.2 系统整体架构怎么搭

我在设计这类系统时，习惯先把数据流画清楚，再决定每一层用什么技术。整个系统的架构可以分成五层，我用一个表格把每层的职责和选型理由说清楚。

这里我要特别说明一下存储层的选择。很多同学一上来就想着上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, threshold=3): 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_news

2.2 Spark集群怎么搭才不踩坑

这是很多同学最头疼的部分。我见过太多人在“Spark集群搭建”这一步卡了一两周，最后放弃改用单机模式。其实对于毕设来说，你不需要一个真正的多节点集群。用Docker搭一个伪分布式环境，或者直接在单机上用local[*]模式跑，完全够用。关键是你要理解Spark的运行机制，而不是真的去维护一个集群。

如果你确实想搭一个多节点集群，我建议用Docker Compose，一台机器上起三个容器：一个Master，两个Worker。这样既能看到分布式的效果，又不用折腾网络配置。内存分配上，每个Worker给2GB，Master给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_MODE=master - SPARK_MASTER_HOST=spark-master ports: - "8080:8080" - "7077:7077" spark-worker-1: image: bitnami/spark:3.5 environment: - SPARK_MODE=worker - SPARK_MASTER_URL=spark://spark-master:7077 - SPARK_WORKER_MEMORY=2G spark-worker-2: image: bitnami/spark:3.5 environment: - SPARK_MODE=worker - SPARK_MASTER_URL=spark://spark-master:7077 - SPARK_WORKER_MEMORY=2G

注意：如果你用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, params=params) if response.status_code == 200: return response.json().get("articles", []) else: print(f"Error: {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(inputCol="content", outputCol="words") hashingTF = HashingTF(inputCol="words", outputCol="rawFeatures", numFeatures=10000) idf = IDF(inputCol="rawFeatures", outputCol="features") lr = LogisticRegression(maxIter=20, regParam=0.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, x="week", y="avg_sentiment", title=f"{stock} 周度情感趋势") st.plotly_chart(fig)

如果你想让可视化更高级，可以做一个热力图，横轴是时间，纵轴是股票，颜色表示情感得分。这样一眼就能看出哪些股票在哪些时间段情绪异常。ECharts的热力图配置稍微复杂一点，但效果很好，网上有很多模板可以直接改。

4. 常见问题与排查技巧实录

4.1 Spark运行中的典型报错与解决

做这个项目，你一定会遇到各种Spark报错。我整理了一个速查表，覆盖了最常见的几个问题。

数据倾斜是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(filename='fetch_errors.log', level=logging.ERROR) def fetch_with_retry(url, params, max_retries=3): for i in range(max_retries): try: response = requests.get(url, params=params, timeout=10) 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(f"Attempt {i+1} failed: {str(e)}") time.sleep(5) return None

还有一个容易被忽略的问题是数据存储的持久化。如果你把数据存在本地，电脑一关机就没了。我建议用云盘同步，或者存到学校的服务器上。如果条件允许，用HDFS是最稳妥的，但配置起来麻烦。折中方案是用SQLite存原始数据，Parquet存中间结果，这样既轻量又可靠。

4.4 毕设答辩时怎么讲清楚技术亮点

最后说一个非技术但很重要的问题：答辩怎么讲。我见过很多同学，代码写得不错，但一上台就只会念PPT，把“我用了Spark”重复十遍。评委想听的不是你用了什么工具，而是你为什么用这个工具，解决了什么问题。

我的建议是，准备三个故事。第一个故事讲数据规模：我抓了多少条新闻，如果用pandas要跑多久，用Spark只用了多久。第二个故事讲技术对比：我试了VADER和逻辑回归，准确率分别是多少，为什么逻辑回归更好。第三个故事讲业务发现：我发现某只股票的情感得分和股价走势有滞后相关性，滞后大概几天。这三个故事讲完，你的毕设就有了灵魂，而不只是一个“跑通的系统”。

另外，答辩时一定要准备一个现场演示。不要只放截图，要真的打开系统，点几下，让评委看到交互效果。如果担心现场网络问题，提前录一个演示视频作为备份。这个细节很多人忽略，但关键时刻能救你一命。

提示：如果你的毕设需要提交代码，记得写一个清晰的README，说明环境依赖、数据来源、运行步骤。评委可能不会跑你的代码，但README的质量直接影响他对你工程能力的判断。我通常会在README里放一个“快速开始”章节，三行命令就能跑起来，这样即使是不懂技术的人也能感受到你的项目是完整的。