简介面向大数据类毕业设计的一份Spark心脏病信息分析源码与数据包内置完整工程结构和配套数据集适合正在完成类似选题的学生、Spark入门者及需要医疗数据分析参考的开发者。包内共1010个文件压缩包约8.93MB核心实现以Scala为主辅以CSV数据文件、JSON配置、Markdown笔记与HTML页面等Scala代码承担数据清洗、特征转换和统计建模逻辑CSV提供心脏病患者诊疗记录配置类文件便于修改运行参数。所有源码均已在本地编译通过、可直接运行项目难度适中内容经过助教老师审定可安全用于课程设计、毕业设计及自学练习。从该工程中可以学到Spark读写CSV、数据预处理、分组聚合、阈值分析等完整分析思路数据与代码分目录存放结构清晰方便按模块逐步复现。目前已有368人浏览学习对医疗健康类大数据项目的搭建具有较强的参考价值。1. 把 Spark 用在心脏病数据上先想清楚第一个问题做毕业设计的人在拿到“基于 Spark 的心脏病信息大数据分析”这个题目时第一反应往往是赶紧打开 IDE把 Spark 环境搭起来跑一个 wordcount 证明自己会用。但真正拉开分数差距的是你能不能回答一个问题为什么这个场景必须要用 Spark而不是 pandas 一把梭常见的数据集比如 UCI Heart Disease 或 Kaggle 上的 heart.csv只有 300 到 1000 条记录规模上完全不算大数据。用 Spark 跑它听起来像是用高射炮打蚊子。但这也是这类题目最有价值的地方它训练的不是“用 Spark 处理大数据”的能力而是“用分布式思维组织数据处理流程”的能力。所以这篇博文不打算只贴一份能交差的源码而是把从数据读入、特征工程、模型训练到参数调优、代码结构组织这一整条链路按毕设答辩时会被追问的顺序讲透。读者可以是正在做类似选题的本科生也可以是刚接触 Spark 想找一份完整数据分析案例的工程师。前 100 字里已经说清楚了一件事这个题目的核心不是算法多深而是你能不能把 Spark 的特性真正用出来而不是把 Spark 当 pandas 用。2. 数据从哪来、长什么样先摸清心脏病数据的“底”2.1 数据集选型UCI 和 Kaggle 上的字段到底怎么对应做毕设最忌讳的是拿到数据不先看 schema直接开始写处理逻辑。Spark 虽然能自动推断类型但自动推断在心脏病数据上经常翻车。以最常用的 heart.csv 为例常见字段有 14 个age年龄、sex性别1 男 0 女、cp胸痛类型取值 0-3、trestbps静息血压、chol胆固醇、fbs空腹血糖是否大于 120mg/dl、restecg静息心电图结果0-2、thalach最大心率、exang运动诱发心绞痛1 是 0 否、oldpeakST 段压低数值、slopeST 段斜率0-2、ca主要血管数0-3、thal地中海贫血取值 1、2、3有时是 0-3、target是否患病1 是 0 否。这套字段有几个坑要注意。ca和thal在原始 UCI 数据里经常出现?作为缺失值而?不是空字符串也不是 null它是字面量问号。如果你用spark.read.csv的默认inferSchema去推断ca列会直接被推断成 string 类型。后面做VectorAssembler时全是数值型字段才能组装这里就会第一个报错。我的习惯是第一次读数据时永远不依赖inferSchema先指定一个宽松的 schema把ca和thal先读取为 string后续再转换后面会写具体代码。2.2 用 SparkSession 读取 CSV先把 schema 钉死直接看代码这一步是把原始 CSV 读进来并打印基本信息的最小可跑片段from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, IntegerType, DoubleType, StringType spark SparkSession.builder \ .appName(HeartDiseaseAnalysis) \ .master(local[*]) \ .config(spark.sql.adaptive.enabled, true) \ .getOrCreate() schema StructType([ StructField(age, IntegerType(), True), StructField(sex, IntegerType(), True), StructField(cp, IntegerType(), True), StructField(trestbps, IntegerType(), True), StructField(chol, IntegerType(), True), StructField(fbs, IntegerType(), True), StructField(restecg, IntegerType(), True), StructField(thalach, IntegerType(), True), StructField(exang, IntegerType(), True), StructField(oldpeak, DoubleType(), True), StructField(slope, IntegerType(), True), StructField(ca, StringType(), True), StructField(thal, StringType(), True), StructField(target, IntegerType(), True) ]) df spark.read \ .option(header, true) \ .option(delimiter, ,) \ .schema(schema) \ .csv(hdfs://namenode:8020/data/heart.csv) df.printSchema() df.show(5, truncateFalse)逻辑说明schema的作用是把字段类型在读取阶段就固定下来避免inferSchema把ca、thal推断成别的类型。csv的schema参数和option(header, true)是配套使用的先指定 schema 再指定路径解析器就会按这个 schema 逐列解析。master(local[*])表示用本机所有核跑这在本地调试阶段是合理选择如果后面要提交到集群这行要么删掉要么改成spark-submit时传入。printSchema()和show(5)是体检动作任何一个 Spark 分析任务里都不该省。前者确认字段名和类型后者直接看值长什么样。这步能一次性暴露出?缺失值、单位不统一、target 是否均衡等问题。2.2.1 处理缺失值问号、空串和 null 是三回事心脏病数据里最经典的坑就是ca和thal两列里面的?。如果你的数据是从 UCI 老版本下载的这个问题百分百存在。处理逻辑我建议这样from pyspark.sql.functions import col, when, count df df.replace(?, None).withColumn( ca, col(ca).cast(IntegerType()) ).withColumn( thal, col(thal).cast(IntegerType()) ) df.select([count(when(col(c).isNull(), c)).alias(c) for c in df.columns]).show()replace(?, None)把问号统一替换成 null然后再 cast 成整数。如果先 cast 后 replace字符串?无法转成整数会换回来 null但日志里会报一堆解析错误。最后一行统计每列的空值数量这个结果应该记下来答辩时老师问你“数据质量问题怎么处理的”你直接把这个表抛出去。对于空值填充常见做法是分列处理。ca用中位数填充thal用众数填充trestbps和chol用均值填充。这里有个细节Spark 的fillna默认按列名匹配你也可以传一个字典。比如df_filled df.fillna( {ca: df.approxQuantile(ca, [0.5], 0.01)[0], thal: df.groupBy(thal).count().orderBy(count, ascendingFalse).first()[0]} )approxQuantile是分布式的分位数近似计算对于大数据集比median快得多。第二个参数的0.01是相对误差越小越准但越慢这个参数在答辩时如果被问到能说出“这是容差”就算过关。3. 用 Spark SQL 与 DataFrame 做特征工程从清洗到可建模3.1 特征工程不是调包先算相关性再决定怎么编码特征工程这一步Spark 提供了两种风格DataFrame API 和 SQL 字符串。两种风格在毕设里最好都出现因为能体现你两种都会。先做一次相关矩阵观察哪些字段强相关from pyspark.ml.feature import VectorAssembler from pyspark.ml.stat import Correlation numeric_cols [age, trestbps, chol, thalach, oldpeak] vec_df VectorAssembler(inputColsnumeric_cols, outputColfeatures).transform(df_filled) corr_matrix Correlation.corr(vec_df, features).collect()[0][0] print(corr_matrix.toString())VectorAssembler把多个数值列组装成一个向量列Correlation.corr在这个向量列上计算皮尔逊相关矩阵。执行完之后你应该能看到oldpeak和thalach与target的相关系数绝对值较大这两个字段在后续建模中占的权重很可能会最高。这个结果要截图作为 EDA 部分的论据。分类字段的处理思路要区分sex、cp、exang、slope、ca、thal都是有顺序或无顺序的类别。cp虽然是 0-3但 0 和 1 之间没有“大 1”的含义它只是四类胸痛所以做成 one-hot 更稳妥。用 Spark 的OneHotEncoder配StringIndexer的写法如下from pyspark.ml.feature import StringIndexer, OneHotEncoder categorical_cols [sex, cp, restecg, exang, slope, ca, thal] indexers [ StringIndexer(inputColcol_name, outputColcol_name _idx, handleInvalidkeep) for col_name in categorical_cols ] encoders [ OneHotEncoder(inputColcol_name _idx, outputColcol_name _vec) for col_name in categorical_cols ]这段代码里有一个关键参数handleInvalidkeep。它的作用是如果新数据里出现训练时没见过的类别不会直接报错而是把这一个样本放到一个特殊的“unknown”桶里。毕设的测试集如果是手工切分的不设置这个参数也能跑但如果你用 CrossValidator 做交叉验证验证集里可能会出现新类别这时候没有handleInvalid任务会直接炸掉。这是 Spark ML 里很多人踩过的坑。数值字段要不要标准化我的建议是如果后面用逻辑回归必须标准化如果用随机森林可以不标准化。因为逻辑回归的梯度下降受量纲影响chol是几百oldpeak是几直接训练会导致 loss 震荡或收敛慢。标准化用StandardScaler配合VectorAssembler做。3.2 DataFrame 和 Spark SQL 的边界什么时候用 SQL 更划算Spark SQL 的createOrReplaceTempView可以注册临时视图让你用 SQL 查询。好处是代码读起来直观项目文档里好写。比如计算患病率按年龄段分布df_filled.createOrReplaceTempView(heart) sql_result spark.sql( SELECT CASE WHEN age 40 THEN under_40 WHEN age BETWEEN 40 AND 55 THEN 40_55 WHEN age BETWEEN 56 AND 70 THEN 56_70 ELSE over_70 END AS age_group, ROUND(AVG(target), 4) AS disease_rate, COUNT(*) AS cnt FROM heart GROUP BY 1 ORDER BY disease_rate DESC ) sql_result.show()这段 SQL 用了CASE WHEN分桶把连续年龄切成四个区间GROUP BY 1表示按第一列分组ROUND(AVG(target), 4)算每个年龄段的患病率。在同一个项目里前面用 DataFrame API 做清洗和特征工程这里用 SQL 做统计探查两种能力都展示了。而且GROUP BY 1这种写法在 Hive 和 Spark SQL 里都支持显得你对 SQL 语法很熟。这里选切 4 个年龄段不是乱来的。医学统计里常见把年龄按十年一档40 岁以下人群在数据集中数量少单独成一档只是为了不干扰 40-55 这个高风险区间的统计。你可以根据自己数据的分布调整断点但在报告里一定要写清楚“为什么这么切”。3.3 构建最终训练集用 Pipeline 把步骤串起来前面是一步步操作到了机器学习阶段就要用Pipeline把特征工程和模型整合成一个对象。这个设计模式在毕设答辩中几乎是必问的它带来的直接好处是模型预测时不需要手动对输入数据做一遍和训练时一模一样的处理Pipeline会把整套流程自动重放一遍。from pyspark.ml import Pipeline from pyspark.ml.feature import StandardScaler from pyspark.ml.classification import LogisticRegression assembler VectorAssembler( inputCols[c _vec for c in categorical_cols] numeric_cols, outputColassembled_features ) scaler StandardScaler(inputColassembled_features, outputColscaled_features) lr LogisticRegression(featuresColscaled_features, labelColtarget, maxIter100) pipeline Pipeline(stagesindexers encoders [assembler, scaler, lr]) train_df, test_df df_filled.randomSplit([0.8, 0.2], seed42) pipeline_model pipeline.fit(train_df)randomSplit的第二个参数seed42是必须写的。如果不固定随机种子你每次跑出来的训练集和测试集都不一样准确率忽高忽低自己都解释不了。seed可以是任意整数但只要固定一次之后的所有复现都一致。注意Pipeline(stages...)这里的 stages 是一个列表列表顺序就是执行顺序。我在代码里把indexers和encoders先展开再追加 assembler、scaler、lr这个顺序是严格的先转索引再 one-hot再组装向量再标准化最后喂给模型。任何一步调换都会报错或导致结果不对。4. 从 MLlib 到调参与评估把模型做得能答辩、讲得清4.1 模型选型逻辑回归打底随机森林做对照毕设里模型不在多而在解释得清楚。最常见也最稳的组合是逻辑回归作为主模型随机森林作为对照组。逻辑回归的输出是概率天然适合医学二分类随机森林能输出特征重要性适合做特征分析。两个模型在评估指标上的对比表是论文和答辩 PPT 里的核心数据。from pyspark.ml.classification import RandomForestClassifier from pyspark.ml.evaluation import BinaryClassificationEvaluator models { lr: pipeline_model.stages[-1], rf: RandomForestClassifier(featuresColscaled_features, labelColtarget, numTrees100, maxDepth6) } rf_pipeline Pipeline(stagesindexers encoders [assembler, scaler, rf_pipeline])这里有个很容易写错的地方随机森林也需要经过 assembler 和 scaler所以你要重新构建一个 pipeline而不是直接拿前面的模型替换。上面的rf_pipeline那行代码精简化写了实际上你可以在前面定义同一个indexers、encoders、assembler、scaler列表然后组成第二个 Pipeline。这样两份管线共享特征工程逻辑代码不重复。4.2 训练完先看混淆矩阵和 AUC别急着报准确率训练完成后评估指标要成套输出。准确率在类别不平衡的数据上会骗人比如患病样本占 70% 时你全预测患病就有 70% 准确率。所以要看 AUC、召回率和精确率医学场景里召回率尤其重要——漏掉一个心脏病患者比误报一个健康人代价更大。from pyspark.ml.evaluation import BinaryClassificationEvaluator pred_df pipeline_model.transform(test_df) evaluator_auc BinaryClassificationEvaluator(labelColtarget, rawPredictionColrawPrediction, metricNameareaUnderROC) auc evaluator_auc.evaluate(pred_df) print(fAUC: {auc:.4f}) # 手动算混淆矩阵 from pyspark.sql.functions import when, col cm pred_df.groupBy(target).pivot( when(col(prediction) 1, pred_1).otherwise(pred_0) ).count().fillna(0) cm.show()混淆矩阵我用groupBy().pivot()手动算比写 UDF 干净。可以看到 pivot 的结果是 target 为行、prediction 为列的四格表。fillna(0)很关键因为类别不平衡时可能存在某一格完全没有样本不填充就是 null后面画热力图就报错。AUC 的评估结果记录在案。常见数据上逻辑回归 AUC 在 0.85-0.92 区间随机森林可能更高一点。如果你的 AUC 低于 0.7优先怀疑特征工程而不是模型——检查是不是把 target 相关的字段泄露进去了比如数据文件里如果有一个字段叫slope且它其实是医生的最终诊断意见那你的模型就是在作弊。4.3 用 ParamGridBuilder 找参数并用 CrossValidator 验证手工调参对于毕设来说也能用但写到报告里说服力不足。Spark ML 的标准做法是ParamGridBuilder建网格CrossValidator做 K 折交叉验证。这里给一个可运行的网格搜索示例from pyspark.ml.tuning import ParamGridBuilder, CrossValidator param_grid ( ParamGridBuilder() .addGrid(lr.regParam, [0.01, 0.1, 0.5]) .addGrid(lr.elasticNetParam, [0.0, 0.5, 1.0]) .addGrid(lr.maxIter, [50, 100]) .build() ) cv CrossValidator( estimatorpipeline, estimatorParamMapsparam_grid, evaluatorevaluator_auc, numFolds5, seed7, parallelism2 ) cv_model cv.fit(train_df) best_model cv_model.bestModelregParam是 L2 正则化系数elasticNetParam是 L1 与 L2 的混合比例0.0表示纯 L21.0表示纯 L1。总共 3×3×2 共 18 组参数5 折交叉验证就是 90 次训练。parallelism2表示同时跑 2 个任务如果集群资源够就调大这个值。这里有个隐蔽的性能问题local[*]模式下parallelism太高会导致线程切换开销大于训练开销所以本地调试时 2 就够了。4.4 特征重要性分析Spark 里怎么读随机森林的结果随机森林训练完特征重要性是一个一维向量直接取出来看rf_stage best_model.stages[-1] importances rf_stage.featureImportances feature_names [c _vec for c in categorical_cols] numeric_cols imp_pairs sorted(zip(feature_names, importances), keylambda x: -x[1]) for name, imp in imp_pairs[:6]: print(f{name}: {imp:.4f})这里featureImportances的顺序对应assembler的inputCols顺序。我之前把 one-hot 后的类别特征排在前面、数值特征排在后面所以feature_names要按同样的顺序拼。排名前三的特征大概率有oldpeak、thalach、cp这与临床知识是一致的——最大心率、ST 段压低和胸痛类型本来就是心脏病诊断的核心指标。5. 让整套代码在集群上跑起来并准备好答辩演示5.1 把本地代码提交到 YARN 集群spark-submit 的正确姿势很多同学的代码在 IDE 里跑得好好的但spark-submit一提交就报各种错。最核心的原因有两点一是路径不对本地用的是file://路径集群上要换成hdfs://路径二是依赖缺失如果用了第三方 Python 包需要在提交时指定。spark-submit \ --master yarn \ --deploy-mode cluster \ --name HeartDiseaseAnalysis \ --queue default \ --executor-memory 4G \ --num-executors 4 \ --executor-cores 2 \ --driver-memory 2G \ --conf spark.sql.shuffle.partitions200 \ --py-files hdfs://namenode:8020/jars/heart_utils.zip \ hdfs://namenode:8020/jars/main.py--py-files用来上传你打包的自定义模块Spark 会把 zip 分发到每个 executor 的 Python 环境里。--conf spark.sql.shuffle.partitions200这个参数很关键默认值是 200如果你的数据量只有几百条shuffle 分区数设为 20 足够否则会产生大量空 task白白浪费调度开销。数据量小的时候建议显式把它调低到与 executor 数量相关比如 4 个 executor 每个 2 核设 20 或 40 都行。另外注意spark-submit提交后日志在 YARN 的 ResourceManager 页面里可以看不要因为 console 没输出就以为没跑起来。cluster模式下 driver 在集群内部输出不会打到本地控制台。5.2 用 checkpoints 和数据缓存减少重复计算清洗后的 DataFrame 如果在代码里复用了多次每次都会重新计算。用cache()可以避免重复读 HDFS用checkpoint()可以把中间结果落盘切断血缘关系。这两个操作在调试阶段格外有用。df_filled df_filled.cache() df_filled.count() # 触发实际计算并缓存 df_filled df_filled.checkpoint(eagerTrue)cache()是懒执行的必须触发一个 action比如count()才会真正缓存。checkpoint(eagerTrue)会立即计算并写入存储目录。Spark 的 checkpoint 默认目录是空值需要先设置spark.sparkContext.setCheckpointDir(hdfs://...)否则会报错。checkpoint 的价值在长任务中更明显它能截断 RDD/DataFrame 的血缘链避免执行计划过长导致的 StackOverflow。5.3 常见报错对照表直接对应错误关键词去查把在日常调试和带学生毕设中遇到的高频报错整理成一张表按错误信息里出现的关键词去匹配解决方向报错关键词出现原因解决思路Unable to instantiate SparkSession没有找到可识别的 master 地址或缺少配置文件检查--master参数本地调试用local[*]Cannot resolve column ca_vecPipeline 中 stages 顺序有误编码步骤没执行检查 stages 列表中StringIndexer/OneHotEncoder是否在VectorAssembler之前Column features must be of type numericVectorAssembler接受了非数值列先printSchema()确认字段类型对ca、thal先 cast 再组装The number of classes is not 1标签列类型是 string 或含非 0/1 值检查target列的取值groupBy(target).count()看一下java.io.IOException: No space left on deviceshuffle 中间数据写满本地磁盘调大spark.local.dir指向多个磁盘目录Py4JJavaError: An error occurred while calling o123.fit数据集为空或所有特征全为 null检查清洗后count()是否为 0这张表的价值在于它不是错题集而是按报错的位置反推问题模式。比如“无法解析列”这类错误90% 是因为 stages 顺序没对上而不是列名拼写错误。下次遇到类似报错先看自己的 stages 顺序。5.4 答辩演示的小经验提前准备三件事最后说三个直接关系到答辩观感的细节。第一准备一张图。把df_filled的字段分布、缺失值统计、相关性热力图、模型 ROC 曲线这四张图放到一份 PDF 里不用复杂的可视化库pandas 加 matplotlib 足够。关键是一边展示一边口述“每一步花了什么时间”因为 Spark UI 里的耗时统计能证明你真的跑过。第二准备好spark-submit的完整命令写在演示文档里现场不用执行但要能逐参数解释。特别是executor-memory和shuffle.partitions老师很爱问“这个参数为什么这么设”你要答得出来不是照抄教程。第三代码注释里写清楚“为什么”而不是“是什么”。比如handleInvalidkeep旁边写一句“防止交叉验证时新类别导致报错”这句注释在评审老师翻代码时留下的印象比你在 PPT 上多放一张架构图更有用。毕业设计的代码结构本来就是给别人读的写法上多用 Pipeline 少用散落的临时变量评审看起来就是整齐的工程代码。本文还有配套的精品资源点击获取
简介面向大数据类毕业设计的一份Spark心脏病信息分析源码与数据包内置完整工程结构和配套数据集适合正在完成类似选题的学生、Spark入门者及需要医疗数据分析参考的开发者。包内共1010个文件压缩包约8.93MB核心实现以Scala为主辅以CSV数据文件、JSON配置、Markdown笔记与HTML页面等Scala代码承担数据清洗、特征转换和统计建模逻辑CSV提供心脏病患者诊疗记录配置类文件便于修改运行参数。所有源码均已在本地编译通过、可直接运行项目难度适中内容经过助教老师审定可安全用于课程设计、毕业设计及自学练习。从该工程中可以学到Spark读写CSV、数据预处理、分组聚合、阈值分析等完整分析思路数据与代码分目录存放结构清晰方便按模块逐步复现。目前已有368人浏览学习对医疗健康类大数据项目的搭建具有较强的参考价值。1. 把 Spark 用在心脏病数据上先想清楚第一个问题做毕业设计的人在拿到“基于 Spark 的心脏病信息大数据分析”这个题目时第一反应往往是赶紧打开 IDE把 Spark 环境搭起来跑一个 wordcount 证明自己会用。但真正拉开分数差距的是你能不能回答一个问题为什么这个场景必须要用 Spark而不是 pandas 一把梭常见的数据集比如 UCI Heart Disease 或 Kaggle 上的 heart.csv只有 300 到 1000 条记录规模上完全不算大数据。用 Spark 跑它听起来像是用高射炮打蚊子。但这也是这类题目最有价值的地方它训练的不是“用 Spark 处理大数据”的能力而是“用分布式思维组织数据处理流程”的能力。所以这篇博文不打算只贴一份能交差的源码而是把从数据读入、特征工程、模型训练到参数调优、代码结构组织这一整条链路按毕设答辩时会被追问的顺序讲透。读者可以是正在做类似选题的本科生也可以是刚接触 Spark 想找一份完整数据分析案例的工程师。前 100 字里已经说清楚了一件事这个题目的核心不是算法多深而是你能不能把 Spark 的特性真正用出来而不是把 Spark 当 pandas 用。2. 数据从哪来、长什么样先摸清心脏病数据的“底”2.1 数据集选型UCI 和 Kaggle 上的字段到底怎么对应做毕设最忌讳的是拿到数据不先看 schema直接开始写处理逻辑。Spark 虽然能自动推断类型但自动推断在心脏病数据上经常翻车。以最常用的 heart.csv 为例常见字段有 14 个age年龄、sex性别1 男 0 女、cp胸痛类型取值 0-3、trestbps静息血压、chol胆固醇、fbs空腹血糖是否大于 120mg/dl、restecg静息心电图结果0-2、thalach最大心率、exang运动诱发心绞痛1 是 0 否、oldpeakST 段压低数值、slopeST 段斜率0-2、ca主要血管数0-3、thal地中海贫血取值 1、2、3有时是 0-3、target是否患病1 是 0 否。这套字段有几个坑要注意。ca和thal在原始 UCI 数据里经常出现?作为缺失值而?不是空字符串也不是 null它是字面量问号。如果你用spark.read.csv的默认inferSchema去推断ca列会直接被推断成 string 类型。后面做VectorAssembler时全是数值型字段才能组装这里就会第一个报错。我的习惯是第一次读数据时永远不依赖inferSchema先指定一个宽松的 schema把ca和thal先读取为 string后续再转换后面会写具体代码。2.2 用 SparkSession 读取 CSV先把 schema 钉死直接看代码这一步是把原始 CSV 读进来并打印基本信息的最小可跑片段from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, IntegerType, DoubleType, StringType spark SparkSession.builder \ .appName(HeartDiseaseAnalysis) \ .master(local[*]) \ .config(spark.sql.adaptive.enabled, true) \ .getOrCreate() schema StructType([ StructField(age, IntegerType(), True), StructField(sex, IntegerType(), True), StructField(cp, IntegerType(), True), StructField(trestbps, IntegerType(), True), StructField(chol, IntegerType(), True), StructField(fbs, IntegerType(), True), StructField(restecg, IntegerType(), True), StructField(thalach, IntegerType(), True), StructField(exang, IntegerType(), True), StructField(oldpeak, DoubleType(), True), StructField(slope, IntegerType(), True), StructField(ca, StringType(), True), StructField(thal, StringType(), True), StructField(target, IntegerType(), True) ]) df spark.read \ .option(header, true) \ .option(delimiter, ,) \ .schema(schema) \ .csv(hdfs://namenode:8020/data/heart.csv) df.printSchema() df.show(5, truncateFalse)逻辑说明schema的作用是把字段类型在读取阶段就固定下来避免inferSchema把ca、thal推断成别的类型。csv的schema参数和option(header, true)是配套使用的先指定 schema 再指定路径解析器就会按这个 schema 逐列解析。master(local[*])表示用本机所有核跑这在本地调试阶段是合理选择如果后面要提交到集群这行要么删掉要么改成spark-submit时传入。printSchema()和show(5)是体检动作任何一个 Spark 分析任务里都不该省。前者确认字段名和类型后者直接看值长什么样。这步能一次性暴露出?缺失值、单位不统一、target 是否均衡等问题。2.2.1 处理缺失值问号、空串和 null 是三回事心脏病数据里最经典的坑就是ca和thal两列里面的?。如果你的数据是从 UCI 老版本下载的这个问题百分百存在。处理逻辑我建议这样from pyspark.sql.functions import col, when, count df df.replace(?, None).withColumn( ca, col(ca).cast(IntegerType()) ).withColumn( thal, col(thal).cast(IntegerType()) ) df.select([count(when(col(c).isNull(), c)).alias(c) for c in df.columns]).show()replace(?, None)把问号统一替换成 null然后再 cast 成整数。如果先 cast 后 replace字符串?无法转成整数会换回来 null但日志里会报一堆解析错误。最后一行统计每列的空值数量这个结果应该记下来答辩时老师问你“数据质量问题怎么处理的”你直接把这个表抛出去。对于空值填充常见做法是分列处理。ca用中位数填充thal用众数填充trestbps和chol用均值填充。这里有个细节Spark 的fillna默认按列名匹配你也可以传一个字典。比如df_filled df.fillna( {ca: df.approxQuantile(ca, [0.5], 0.01)[0], thal: df.groupBy(thal).count().orderBy(count, ascendingFalse).first()[0]} )approxQuantile是分布式的分位数近似计算对于大数据集比median快得多。第二个参数的0.01是相对误差越小越准但越慢这个参数在答辩时如果被问到能说出“这是容差”就算过关。3. 用 Spark SQL 与 DataFrame 做特征工程从清洗到可建模3.1 特征工程不是调包先算相关性再决定怎么编码特征工程这一步Spark 提供了两种风格DataFrame API 和 SQL 字符串。两种风格在毕设里最好都出现因为能体现你两种都会。先做一次相关矩阵观察哪些字段强相关from pyspark.ml.feature import VectorAssembler from pyspark.ml.stat import Correlation numeric_cols [age, trestbps, chol, thalach, oldpeak] vec_df VectorAssembler(inputColsnumeric_cols, outputColfeatures).transform(df_filled) corr_matrix Correlation.corr(vec_df, features).collect()[0][0] print(corr_matrix.toString())VectorAssembler把多个数值列组装成一个向量列Correlation.corr在这个向量列上计算皮尔逊相关矩阵。执行完之后你应该能看到oldpeak和thalach与target的相关系数绝对值较大这两个字段在后续建模中占的权重很可能会最高。这个结果要截图作为 EDA 部分的论据。分类字段的处理思路要区分sex、cp、exang、slope、ca、thal都是有顺序或无顺序的类别。cp虽然是 0-3但 0 和 1 之间没有“大 1”的含义它只是四类胸痛所以做成 one-hot 更稳妥。用 Spark 的OneHotEncoder配StringIndexer的写法如下from pyspark.ml.feature import StringIndexer, OneHotEncoder categorical_cols [sex, cp, restecg, exang, slope, ca, thal] indexers [ StringIndexer(inputColcol_name, outputColcol_name _idx, handleInvalidkeep) for col_name in categorical_cols ] encoders [ OneHotEncoder(inputColcol_name _idx, outputColcol_name _vec) for col_name in categorical_cols ]这段代码里有一个关键参数handleInvalidkeep。它的作用是如果新数据里出现训练时没见过的类别不会直接报错而是把这一个样本放到一个特殊的“unknown”桶里。毕设的测试集如果是手工切分的不设置这个参数也能跑但如果你用 CrossValidator 做交叉验证验证集里可能会出现新类别这时候没有handleInvalid任务会直接炸掉。这是 Spark ML 里很多人踩过的坑。数值字段要不要标准化我的建议是如果后面用逻辑回归必须标准化如果用随机森林可以不标准化。因为逻辑回归的梯度下降受量纲影响chol是几百oldpeak是几直接训练会导致 loss 震荡或收敛慢。标准化用StandardScaler配合VectorAssembler做。3.2 DataFrame 和 Spark SQL 的边界什么时候用 SQL 更划算Spark SQL 的createOrReplaceTempView可以注册临时视图让你用 SQL 查询。好处是代码读起来直观项目文档里好写。比如计算患病率按年龄段分布df_filled.createOrReplaceTempView(heart) sql_result spark.sql( SELECT CASE WHEN age 40 THEN under_40 WHEN age BETWEEN 40 AND 55 THEN 40_55 WHEN age BETWEEN 56 AND 70 THEN 56_70 ELSE over_70 END AS age_group, ROUND(AVG(target), 4) AS disease_rate, COUNT(*) AS cnt FROM heart GROUP BY 1 ORDER BY disease_rate DESC ) sql_result.show()这段 SQL 用了CASE WHEN分桶把连续年龄切成四个区间GROUP BY 1表示按第一列分组ROUND(AVG(target), 4)算每个年龄段的患病率。在同一个项目里前面用 DataFrame API 做清洗和特征工程这里用 SQL 做统计探查两种能力都展示了。而且GROUP BY 1这种写法在 Hive 和 Spark SQL 里都支持显得你对 SQL 语法很熟。这里选切 4 个年龄段不是乱来的。医学统计里常见把年龄按十年一档40 岁以下人群在数据集中数量少单独成一档只是为了不干扰 40-55 这个高风险区间的统计。你可以根据自己数据的分布调整断点但在报告里一定要写清楚“为什么这么切”。3.3 构建最终训练集用 Pipeline 把步骤串起来前面是一步步操作到了机器学习阶段就要用Pipeline把特征工程和模型整合成一个对象。这个设计模式在毕设答辩中几乎是必问的它带来的直接好处是模型预测时不需要手动对输入数据做一遍和训练时一模一样的处理Pipeline会把整套流程自动重放一遍。from pyspark.ml import Pipeline from pyspark.ml.feature import StandardScaler from pyspark.ml.classification import LogisticRegression assembler VectorAssembler( inputCols[c _vec for c in categorical_cols] numeric_cols, outputColassembled_features ) scaler StandardScaler(inputColassembled_features, outputColscaled_features) lr LogisticRegression(featuresColscaled_features, labelColtarget, maxIter100) pipeline Pipeline(stagesindexers encoders [assembler, scaler, lr]) train_df, test_df df_filled.randomSplit([0.8, 0.2], seed42) pipeline_model pipeline.fit(train_df)randomSplit的第二个参数seed42是必须写的。如果不固定随机种子你每次跑出来的训练集和测试集都不一样准确率忽高忽低自己都解释不了。seed可以是任意整数但只要固定一次之后的所有复现都一致。注意Pipeline(stages...)这里的 stages 是一个列表列表顺序就是执行顺序。我在代码里把indexers和encoders先展开再追加 assembler、scaler、lr这个顺序是严格的先转索引再 one-hot再组装向量再标准化最后喂给模型。任何一步调换都会报错或导致结果不对。4. 从 MLlib 到调参与评估把模型做得能答辩、讲得清4.1 模型选型逻辑回归打底随机森林做对照毕设里模型不在多而在解释得清楚。最常见也最稳的组合是逻辑回归作为主模型随机森林作为对照组。逻辑回归的输出是概率天然适合医学二分类随机森林能输出特征重要性适合做特征分析。两个模型在评估指标上的对比表是论文和答辩 PPT 里的核心数据。from pyspark.ml.classification import RandomForestClassifier from pyspark.ml.evaluation import BinaryClassificationEvaluator models { lr: pipeline_model.stages[-1], rf: RandomForestClassifier(featuresColscaled_features, labelColtarget, numTrees100, maxDepth6) } rf_pipeline Pipeline(stagesindexers encoders [assembler, scaler, rf_pipeline])这里有个很容易写错的地方随机森林也需要经过 assembler 和 scaler所以你要重新构建一个 pipeline而不是直接拿前面的模型替换。上面的rf_pipeline那行代码精简化写了实际上你可以在前面定义同一个indexers、encoders、assembler、scaler列表然后组成第二个 Pipeline。这样两份管线共享特征工程逻辑代码不重复。4.2 训练完先看混淆矩阵和 AUC别急着报准确率训练完成后评估指标要成套输出。准确率在类别不平衡的数据上会骗人比如患病样本占 70% 时你全预测患病就有 70% 准确率。所以要看 AUC、召回率和精确率医学场景里召回率尤其重要——漏掉一个心脏病患者比误报一个健康人代价更大。from pyspark.ml.evaluation import BinaryClassificationEvaluator pred_df pipeline_model.transform(test_df) evaluator_auc BinaryClassificationEvaluator(labelColtarget, rawPredictionColrawPrediction, metricNameareaUnderROC) auc evaluator_auc.evaluate(pred_df) print(fAUC: {auc:.4f}) # 手动算混淆矩阵 from pyspark.sql.functions import when, col cm pred_df.groupBy(target).pivot( when(col(prediction) 1, pred_1).otherwise(pred_0) ).count().fillna(0) cm.show()混淆矩阵我用groupBy().pivot()手动算比写 UDF 干净。可以看到 pivot 的结果是 target 为行、prediction 为列的四格表。fillna(0)很关键因为类别不平衡时可能存在某一格完全没有样本不填充就是 null后面画热力图就报错。AUC 的评估结果记录在案。常见数据上逻辑回归 AUC 在 0.85-0.92 区间随机森林可能更高一点。如果你的 AUC 低于 0.7优先怀疑特征工程而不是模型——检查是不是把 target 相关的字段泄露进去了比如数据文件里如果有一个字段叫slope且它其实是医生的最终诊断意见那你的模型就是在作弊。4.3 用 ParamGridBuilder 找参数并用 CrossValidator 验证手工调参对于毕设来说也能用但写到报告里说服力不足。Spark ML 的标准做法是ParamGridBuilder建网格CrossValidator做 K 折交叉验证。这里给一个可运行的网格搜索示例from pyspark.ml.tuning import ParamGridBuilder, CrossValidator param_grid ( ParamGridBuilder() .addGrid(lr.regParam, [0.01, 0.1, 0.5]) .addGrid(lr.elasticNetParam, [0.0, 0.5, 1.0]) .addGrid(lr.maxIter, [50, 100]) .build() ) cv CrossValidator( estimatorpipeline, estimatorParamMapsparam_grid, evaluatorevaluator_auc, numFolds5, seed7, parallelism2 ) cv_model cv.fit(train_df) best_model cv_model.bestModelregParam是 L2 正则化系数elasticNetParam是 L1 与 L2 的混合比例0.0表示纯 L21.0表示纯 L1。总共 3×3×2 共 18 组参数5 折交叉验证就是 90 次训练。parallelism2表示同时跑 2 个任务如果集群资源够就调大这个值。这里有个隐蔽的性能问题local[*]模式下parallelism太高会导致线程切换开销大于训练开销所以本地调试时 2 就够了。4.4 特征重要性分析Spark 里怎么读随机森林的结果随机森林训练完特征重要性是一个一维向量直接取出来看rf_stage best_model.stages[-1] importances rf_stage.featureImportances feature_names [c _vec for c in categorical_cols] numeric_cols imp_pairs sorted(zip(feature_names, importances), keylambda x: -x[1]) for name, imp in imp_pairs[:6]: print(f{name}: {imp:.4f})这里featureImportances的顺序对应assembler的inputCols顺序。我之前把 one-hot 后的类别特征排在前面、数值特征排在后面所以feature_names要按同样的顺序拼。排名前三的特征大概率有oldpeak、thalach、cp这与临床知识是一致的——最大心率、ST 段压低和胸痛类型本来就是心脏病诊断的核心指标。5. 让整套代码在集群上跑起来并准备好答辩演示5.1 把本地代码提交到 YARN 集群spark-submit 的正确姿势很多同学的代码在 IDE 里跑得好好的但spark-submit一提交就报各种错。最核心的原因有两点一是路径不对本地用的是file://路径集群上要换成hdfs://路径二是依赖缺失如果用了第三方 Python 包需要在提交时指定。spark-submit \ --master yarn \ --deploy-mode cluster \ --name HeartDiseaseAnalysis \ --queue default \ --executor-memory 4G \ --num-executors 4 \ --executor-cores 2 \ --driver-memory 2G \ --conf spark.sql.shuffle.partitions200 \ --py-files hdfs://namenode:8020/jars/heart_utils.zip \ hdfs://namenode:8020/jars/main.py--py-files用来上传你打包的自定义模块Spark 会把 zip 分发到每个 executor 的 Python 环境里。--conf spark.sql.shuffle.partitions200这个参数很关键默认值是 200如果你的数据量只有几百条shuffle 分区数设为 20 足够否则会产生大量空 task白白浪费调度开销。数据量小的时候建议显式把它调低到与 executor 数量相关比如 4 个 executor 每个 2 核设 20 或 40 都行。另外注意spark-submit提交后日志在 YARN 的 ResourceManager 页面里可以看不要因为 console 没输出就以为没跑起来。cluster模式下 driver 在集群内部输出不会打到本地控制台。5.2 用 checkpoints 和数据缓存减少重复计算清洗后的 DataFrame 如果在代码里复用了多次每次都会重新计算。用cache()可以避免重复读 HDFS用checkpoint()可以把中间结果落盘切断血缘关系。这两个操作在调试阶段格外有用。df_filled df_filled.cache() df_filled.count() # 触发实际计算并缓存 df_filled df_filled.checkpoint(eagerTrue)cache()是懒执行的必须触发一个 action比如count()才会真正缓存。checkpoint(eagerTrue)会立即计算并写入存储目录。Spark 的 checkpoint 默认目录是空值需要先设置spark.sparkContext.setCheckpointDir(hdfs://...)否则会报错。checkpoint 的价值在长任务中更明显它能截断 RDD/DataFrame 的血缘链避免执行计划过长导致的 StackOverflow。5.3 常见报错对照表直接对应错误关键词去查把在日常调试和带学生毕设中遇到的高频报错整理成一张表按错误信息里出现的关键词去匹配解决方向报错关键词出现原因解决思路Unable to instantiate SparkSession没有找到可识别的 master 地址或缺少配置文件检查--master参数本地调试用local[*]Cannot resolve column ca_vecPipeline 中 stages 顺序有误编码步骤没执行检查 stages 列表中StringIndexer/OneHotEncoder是否在VectorAssembler之前Column features must be of type numericVectorAssembler接受了非数值列先printSchema()确认字段类型对ca、thal先 cast 再组装The number of classes is not 1标签列类型是 string 或含非 0/1 值检查target列的取值groupBy(target).count()看一下java.io.IOException: No space left on deviceshuffle 中间数据写满本地磁盘调大spark.local.dir指向多个磁盘目录Py4JJavaError: An error occurred while calling o123.fit数据集为空或所有特征全为 null检查清洗后count()是否为 0这张表的价值在于它不是错题集而是按报错的位置反推问题模式。比如“无法解析列”这类错误90% 是因为 stages 顺序没对上而不是列名拼写错误。下次遇到类似报错先看自己的 stages 顺序。5.4 答辩演示的小经验提前准备三件事最后说三个直接关系到答辩观感的细节。第一准备一张图。把df_filled的字段分布、缺失值统计、相关性热力图、模型 ROC 曲线这四张图放到一份 PDF 里不用复杂的可视化库pandas 加 matplotlib 足够。关键是一边展示一边口述“每一步花了什么时间”因为 Spark UI 里的耗时统计能证明你真的跑过。第二准备好spark-submit的完整命令写在演示文档里现场不用执行但要能逐参数解释。特别是executor-memory和shuffle.partitions老师很爱问“这个参数为什么这么设”你要答得出来不是照抄教程。第三代码注释里写清楚“为什么”而不是“是什么”。比如handleInvalidkeep旁边写一句“防止交叉验证时新类别导致报错”这句注释在评审老师翻代码时留下的印象比你在 PPT 上多放一张架构图更有用。毕业设计的代码结构本来就是给别人读的写法上多用 Pipeline 少用散落的临时变量评审看起来就是整齐的工程代码。本文还有配套的精品资源点击获取
)
)
