先说个真实场景。上学期期末同事拉着一张成绩明细表来找我十三个班、六门课、加上历次补考记录将近二十万行。他原来的做法是用Excel按班级筛出来挨个算一份报告折腾两天。我当时看了一眼文件第一反应是用Pandas跑但再一问后面还要做年级排名、学科均衡度、进退步追踪这些活放在单机内存里跑一遍可以反复调条件再跑就不划算了。于是我把整条分析链路放到了Spark上从清洗、聚合、排名到波动追踪全程用DataFrame和SQL跑通。这篇文章就把这个项目从数据形态、Schema设计、清洗逻辑、分析模型到调优排坑完整复盘一遍适合手里握着几十万行成绩表、不想再被Excel和SQL来回折腾的读者参考。1. 几十万行成绩数据为什么要动Spark1.1 一次期末成绩分析暴露的单机瓶颈同事原来处理成绩的方式是典型的“先筛后算”按班主任要一份按年级要一份按科目要一份每个都要重新打开原始表用透视表刷一次。数据量到了二十万行之后Excel打开文件明显吃内存筛选时转圈做一次透视能去倒杯水。我最初也觉得这量级Pandas完全能扛因为二十万行DataFrame在本地跑均值方差并不慢真正让我改主意的是分析任务的数量和迭代方式。成绩分析的痛点从来不是“算一次”而是“换条件反复算”。今天算全校平均分明天按班级加科目后天再要连续三学期的对比每来一个新需求Pandas脚本就要改一段重新跑。整个过程对临时变量和内存的复用很差而且随着学期数据累积半年一年后轻松突破百万行单机内存就会开始冒汗。1.2 Spark在这个场景里的三个核心优势我最后选择Spark不是因为大数据四个字听起来唬人而是它的三个特性恰好打在成绩分析的需求点上懒计算与执行计划Spark的DataFrame是延迟执行的只有真正触发action才跑任务。同一个清洗流程定义好之后后续按不同维度groupBy、join、filterSpark会重新生成最优执行计划不用像Pandas那样关心中间结果的物理存储。同一套逻辑兼容SQL与API我对复杂逻辑用SQL表达顺手对循环和窗口操作用API更直观Spark两边都能写且性能上没差别因为最终都会转成逻辑计划。扩容简单单机跑不动时从local模式改成standalone或者YARN代码一行不用改资源往上加就行。这一点在后续数据量继续膨胀时特别值钱。1.3 我的选型判断标准这不是鼓吹任何场景都该用Spark。如果只是几百行成绩Excel和Pandas仍然是效率最高的方案。我更倾向于用这样几条标准来判断数据量是否在几十万行以上且还在持续累积分析需求是否频繁变化需要同一套数据多次加工是否需要跨表join、窗口排名、按班级/科目/学期反复透视这类多种计算混合是否希望处理流程沉淀下来下次数据来了自动跑一遍。四个条件满足一半以上Spark的优势就明显了。这次项目四条全占所以没有犹豫。2. 开工前的数据形态与Schema设计2.1 成绩数据最常见的两种长相做这个项目之前我先整理了成绩数据的原始形态。教务系统导出的数据大致分成两种长表一行一个学生的一门科目成绩字段通常是学号、姓名、班级、科目、分数、考试名称、考试日期。适合做聚合、筛选和排名。宽表一行一个学生各科目成绩横向展开适合看个体画像、算学科均衡度。这次从教务系统拿到的是长表CSV格式文件大约180MB包含七个核心字段。我一开始没急着分析而是先确认了每个字段的格式尤其是学号有的导出文件里学号带前导零一旦被当成数字读进来就全丢了。2.2 PySpark环境搭建与提交方式取舍环境方面我用了PySpark 3.x。开发阶段用Jupyter连本地SparkSession调试跑通之后用spark-submit提交正式作业日志和依赖都更可控。提交命令大致是这样spark-submit \ --master yarn \ --deploy-mode cluster \ --executor-memory 4g \ --driver-memory 2g \ --num-executors 8 \ analyze_scores.py有个细节值得单独说本地调试和集群提交的资源参数不要拍脑袋。我第一版直接把executor内存开到8g结果队列资源不够任务一直等。后面改成4g、8个executor总资源和单任务需求都平衡了。2.3 手动定义Schema比inferSchema靠谱得多读CSV时最容易踩的坑就是让Spark自动推断类型。inferSchema看着省事实际隐患很多。成绩表里如果有空值某个全是空值的列会被推断成string带前导零的学号ID可能被推断成bigint清洗阶段所有ID都对不上人。而且全表扫描推断类型需要额外跑一遍数据文件大了很浪费时间。这次我直接手动定义了Schemafrom pyspark.sql.types import StructType, StructField, StringType, DoubleType, DateType score_schema StructType([ StructField(student_id, StringType(), False), StructField(student_name, StringType(), True), StructField(class_name, StringType(), True), StructField(subject, StringType(), True), StructField(score, DoubleType(), True), StructField(exam_name, StringType(), True), StructField(exam_date, DateType(), True) ]) df spark.read \ .option(header, True) \ .option(delimiter, ,) \ .schema(score_schema) \ .csv(hdfs://path/to/score_data/)手动定义Schema有几个明确好处类型对了空值处理规则可预期加载时跳过推断扫描执行计划更早定型。对成绩这种字段固定的结构化数据完全没有必要让Spark猜类型。3. 清洗环节成绩表比想象中脏得多3.1 缺失值、重复行与无效记录的判断依据成绩表的脏数据问题主要集中在这几类处理时各有逻辑学号缺失或学号格式异常学号是后续所有join的关联键缺失了就直接丢或标记为“待核对”不能留到分析环节。成绩为空很多补考记录只有参考情况没有实际分数。这类如果做及格率统计不能直接丢掉要区分“缺考”和“低分”我会额外加标记字段而不是简单过滤。明显重复行同一学号科目考试名称完全重复的记录大概率是导出时数据源join查重没做好dropDuplicates可以解决。要注意dropDuplicates不是无脑全字段去重必须指定子集。否则两个人同名同分可能整行完全一样就会被误删。正确的做法是df_clean df.dropDuplicates([student_id, subject, exam_name, exam_date])3.2 异常分数识别百分制下的合理范围分数异常相对好处理。百分制成绩理论上应该在0到100之间但实际看数据时发现两类极端情况一是总分不是百分制比如某些科目满分150直接用固定阈值会误杀二是录入时小数点位置出错比如把88.5录成885。我的处理思路是按科目满分表join一个映射字段再统一校验。先给成绩表加一列full_score再做规则判断from pyspark.sql.functions import when, col, lit df_validated df_clean.withColumn( is_valid, when((col(score) 0) (col(score) col(full_score)), lit(1)).otherwise(lit(0)) ) invalid_scores df_validated.filter(col(is_valid) 0)异常分数不要直接删。先把异常行单独导出给教务核对因为有些是录入错误后续修正后还能重新合入主表。盲删会把本来可以救回来的数据丢掉。3.3 长表转宽表什么时候必须转分析过程中长表适合绝大多数聚合和排名场景但做学生个人多科对比和学科均衡度时长表写起来很绕。比如想判断一个学生六门课里哪门最弱最直观的是把六科变成六列然后横向算分差。Spark的pivot操作正好解决这个问题wide_df df_validated \ .groupBy(student_id, student_name, class_name, exam_name) \ .pivot(subject) \ .agg(agg_first(score))需要注意pivot聚合函数要选对。同一个学生在同一场考试里每个科目只会有一条记录用first没问题但如果数据没去干净存在多科多条记录first可能拿到任意一条产生统计偏差。所以我做pivot前会再做一次子集去重确保每个student_idsubjectexam_name唯一。4. 核心分析链路聚合、排名、科目对比4.1 全校与班级维度的汇总统计基础的统计量计算直接用agg一次搞定。我用一条SQL表达了“各科目各班次的平均值、标准差、最高最低分、考试人次、及格率”这种聚合是成绩分析的高频操作SQL读起来比链式API更直观SELECT subject, class_name, COUNT(score) AS exam_cnt, ROUND(AVG(score), 2) AS avg_score, ROUND(STDDEV(score), 2) AS std_score, MAX(score) AS max_score, MIN(score) AS min_score, ROUND(SUM(CASE WHEN score 60 THEN 1 ELSE 0 END) / COUNT(score), 4) AS pass_rate FROM cleaned_scores GROUP BY subject, class_name ORDER BY subject, avg_score DESC这里及格率我直接写在SQL里用CASE WHEN转布尔值再取均值逻辑清楚。如果后续想把及格线从60改成不同科目的标准只需要改这个条件表达式不用动整体结构。4.2 窗口函数做排名row_number和dense_rank的区别排名是教务老师最常问的需求年级排名、班级排名、单科排名。Spark窗口函数处理这类问题非常顺手from pyspark.sql.window import Window from pyspark.sql.functions import row_number, dense_rank, rank window_rank Window.partitionBy(class_name, subject).orderBy(col(score).desc()) ranked_df df_validated \ .withColumn(rank_in_class, row_number().over(window_rank)) \ .withColumn(dense_rank_in_class, dense_rank().over(window_rank))这里必须说清楚几个排序函数的差异。row_number不管成绩是否相同都会给一个唯一序号rank遇到并列名次会跳号比如两个并列第二下一个人排第四dense_rank并列不跳号两个并列第二后下一个人排第三。做奖励名单时通常用dense_rank做数据导出定位唯一行时用row_number。这个项目里我两种都生成了后续使用场景不同。用整数排名做唯一标识用dense_rank做“前多少名”筛选。4.3 科目横向对比与均衡度指标除了排名教务更想看到的是“哪些班哪科明显偏弱”。长表里可以直接按科目分组算平均分排出名次但要分析“一个班六科是否均衡”就需要把班×科的平均分矩阵算出来subject_pivot df_validated \ .groupBy(class_name) \ .pivot(subject) \ .agg(round(avg(score), 2))拿到宽表之后我可以横向计算一个学生或一个班级的多科标准差。标准差越大说明各科发展越不均衡。更专业的做法是算变异系数也就是标准差除以均值消除不同科目满分不同的影响。在成绩场景里变异系数超过0.2基本就能认定偏科明显需要重点关注。5. 进阶分析成绩波动追踪与群体分层5.1 两次考试之间的进退步分析单纯的静态排名说服力有限教务真正关心的是变化趋势。我把同一学生同一科目两次考试做自连接计算分差SELECT a.student_id, a.student_name, a.class_name, a.subject, a.score AS current_score, b.score AS previous_score, ROUND(a.score - b.score, 2) AS score_diff FROM current_exam a LEFT JOIN previous_exam b ON a.student_id b.student_id AND a.subject b.subject WHERE a.exam_name 2024年秋季期末 AND b.exam_name 2024年秋季期中 ORDER BY score_diff ASC这里的join键是student_id subject刚好是唯一的。分差排序后退步最大的学生自然在最前面。我随后接了一个阈值筛选分差低于-15的单独导出一份名单交给班主任逐个核对原因。 这种纵向分析在Excel里做很麻烦但Spark的分布式join加上谓词下推跑起来非常快数据量大时优势更明显。5.2 分位数分层把学生群体切成五层除了看个体我还需要做群体分层方便教务掌握整体分布。用ntile窗口函数把学生按总成绩分成五等份layer_window Window.partitionBy(class_name).orderBy(col(total_score).desc()) layered_df student_total_df.withColumn( score_level, ntile(5).over(layer_window) )分层之后配合各层人数的交叉表可以直观看到每个班在不同分数段的人数分布。这里有个细节ntile是按当前分区的行数平均分桶如果某班人数不是5的倍数靠后的桶人数会略少这是正常现象。做四层或者五层对教学分析来说差异不大关键在于层数要固定方便不同班级之间横向比较。5.3 连续多学期成绩的纵向累积分析单个学期只是快照把连续四个学期的数据叠加起来能看出更长期的趋势。我把多个学期的成绩表纵向union到一起再按学期分组算班级平均分。如果一个班级的班级平均分连续两个学期下滑那就不只是某个学生的问题可能需要从教学安排上找原因。这类纵向累积分析在Spark里就是加一个unionByName再接聚合没有额外的复杂度。结合这些分析结果最终输出给教务的包括各科平均分、及格率、标准差汇总表班级排名和年级并列名次列表单科波动最大的前20名和后20名按五层划分的各班人数分布交叉表连续三学期班级平均分趋势表。所有结果统一以分区CSV写入指定目录教务直接下载就能用。6. 实际运行中的调优与排坑经验6.1 缓存策略什么时候cache什么时候别乱cache成绩分析任务里同一份清洗后的DataFrame会被反复使用。第一次跑我在每个聚合前都重新读了一遍源数据程序慢但还能忍受。后来加了多个分析步骤每个action都从头开始做一遍全量清洗日志里看到的读文件时间越来越离谱。这时候就该cache()了df_validated.cache() df_validated.count() # 触发实际缓存cache()不是一调用就生效必须遇到action才会真正把数据放到内存里。上面的count()就是为了触发缓存。缓存之后后续多次groupBy、join、pivot都会明显变快。但cache()也不是越多越好。如果数据本身很小或者只用一次就丢缓存反而浪费内存。这种场景用checkpoint或者干脆不做更合适。成绩数据一般在几百MB到几GB之间缓存收益最大。6.2 Shuffle分区数默认200未必合适Spark的spark.sql.shuffle.partitions默认是200这是为较大集群准备的。我这次处理的数据量不算特别大按默认200个shuffle分区跑每个分区只有几万条记录任务启动和序列化的开销比计算本身还大。我把参数调成了40作业耗时立刻下来一大截spark-submit \ --conf spark.sql.shuffle.partitions40 \ --conf spark.default.parallelism40 \ analyze_scores.py具体值怎么定可以先用df.rdd.getNumPartitions()看当前分区数再结合spark.sql.shuffle.partitions调整。一个经验区间是每个分区控制在100MB到200MB之间这样单任务处理时间既不会太长也不会因为分区太多而产生过多调度开销。6.3 数据倾斜问题在这里的真实形态成绩数据很少出现传统意义上的“热点Key倾斜”但还是有一种隐蔽的倾斜某个科目的参与人数远高于其他科目。比如“语文”这类全员必考科目在按科目做join时对应的数据量就是“艺术”类选修课的好几倍。我在做“当前成绩对比历史成绩”的join时一开始没有注意到这一点跑起来发现某些stage明显比其他stage慢拖尾严重。日志里看到个别task处理的数据量比平均值大了七八倍。缓解办法是在join前先按科目做过滤只保留需要对比的科目范围同时把经常一起参与计算的列提前做repartition。6.4 小文件问题导出结果时最容易忽略的坑分析完成后我把所有结果用df.write.csv()直接写出结果发现每个结果目录下生成了一堆几十KB的小文件。教务老师拿去一打开Excel卡得厉害而且往Hive里加载这种目录也会产生严重的小文件问题。处理办法是在写出前做一次coalesce把分区合并到合适的数量result_df.coalesce(4).write.mode(overwrite).option(header, True).csv(output_path)coalesce(n)会把分区数减少到n从而减少输出文件数量。这里不建议用repartition因为repalition会触发一次全量shuffle而coalesce能尽量在现有分区上合并成本低得多。6.5 几个容易写错的API细节最后补几个这次项目里真实踩过的API问题pivot之前必须确保唯一性否则聚合结果不可控when和case when的条件顺序要小心多条件时要显式用括号包住重叠条件日期字段在CSV里是字符串手动定义Schema为DateType后如果源文件格式不标准读进来会全部变成null所以日期字段我会先保留StringType清洗阶段再统一转orderBy窗口函数默认升序做排名时desc必须显式写出漏掉一次就会变成“最低分排名第一”那种结果发到工作群里很难收场。这些坑单独看都不大但组合在一起足以让一个Spark作业多跑大半天。这次把整套成绩分析流程迁到Spark之后我的体感是分析代码的复用性变强了下个学期数据导进来只要Schema没变清洗和聚合链路直接跑就行不用再对着Excel反复刷透视表。过程中踩过的那些坑多半不是Spark本身的问题而是对数据形态和API细节想得太简单。如果你也在用Spark处理类似的结构化表格数据希望这篇复盘能帮你把流程走顺。