简介这份资源面向高校大数据教学、课程设计及数据分析入门者提供一套基于Spark与Scala、集成Hive数据仓库的完整学生行为分析项目。项目围绕一卡通消费记录、图书借阅数据与图书馆门禁日志展开多维度清洗预处理并借助KMeans聚类算法划分学生消费水平与生活规律可用于校园行为洞察、个性化服务设计等场景。压缩包共67个文件约7.15MB以15个Scala源码文件为核心辅以Java代码、XML与properties配置、txt说明及md文档另含测试用例与示例数据集结构完整便于直接运行与二次开发。目前已有57人学习下载。读者可从中获取从数据清洗、Hive建表查询到聚类建模的完整实现链路理解Spark与Scala协同处理大规模数据的工程组织方式并参考其目录结构与配置思路快速搭建自己的校园数据分析实验环境。1. 高校一卡通数据治理从 Spark 到 KMeans 的完整链路为什么值得做高校一卡通系统每天产生的数据远比想象中杂食堂刷卡、超市消费、图书馆借还书、门禁进出四类日志分散在不同业务库字段命名、时间格式、主键规则全都不一样。想拿这些数据做学生消费水平与生活规律的聚类分析第一步不是调 KMeans而是先把 Spark、Scala、Hive 这条链路搭稳把脏数据洗成能进模型的特征表。我带过的几个校园数据项目里真正吃掉七成工时的从来不是聚类算法本身而是清洗预处理——空值、重复刷卡、跨天消费、借阅记录与门禁时间对不齐这些坑不填平KMeans 跑出来的簇就是玄学。这篇笔记面向有 Scala 和 SQL 基础、想用 Spark 集成 Hive 做校园行为分析的同学从环境、清洗、特征工程一路写到 KMeans 调参和结果验证每一步都给可复现的命令和参数熟手可以直接跳到避坑章节看边界。2. Spark 集成 Hive 的环境搭建与数据接入2.1 为什么选 Spark on Hive 而不是纯 Hive SQL纯 Hive SQL 做多表关联和窗口统计没问题但一卡通数据要做迭代式特征工程——反复试不同的时间窗口、不同的聚合维度Hive 每次都要落盘 MapReduce一轮下来十几分钟。Spark 把中间结果放内存同样的清洗逻辑用 Scala DataFrame API 写迭代速度差一个数量级。更关键的是 KMeans 在 Spark MLlib 里是原生算子特征表直接从 Hive 读进来就能喂给模型不用在 Hive 和 Python 之间来回导数据。选型上我一般这么定原始日志的落地和分区管理交给 Hive保证元数据和分区裁剪清洗、特征拼接、聚类全部在 Spark 里做。Spark 通过enableHiveSupport()直接读写 Hive 表元数据走同一个 Metastore两边看到的是同一份表结构不会出现「Hive 里改了字段 Spark 读不到」的翻车。2.2 环境搭建的最小可用配置集群规模不大的话三节点就够跑通全流程一个 Master 兼 Metastore两个 Worker。Spark 用 3.x 版本Scala 用 2.12Hive 用 3.x三者版本要对齐Scala 2.13 和 Spark 3.0 以下是不兼容的这是最常见的启动报错来源。# 解压并配置 Spark 环境变量 tar -zxvf spark-3.3.0-bin-hadoop3.tgz -C /opt/ export SPARK_HOME/opt/spark-3.3.0-bin-hadoop3 export PATH$SPARK_HOME/bin:$PATH # 把 Hive 的配置文件软链到 Spark conf 目录让 Spark 找到 Metastore ln -s /opt/hive/conf/hive-site.xml $SPARK_HOME/conf/hive-site.xml # 启动 Spark 集群Standalone 模式 $SPARK_HOME/sbin/start-all.sh # 验证 Spark 能读到 Hive 表 $SPARK_HOME/bin/spark-sql -e show databases;这段配置的核心是hive-site.xml的软链。Spark 默认用自己的内置 Metastore如果不指向 Hive 的配置spark-sql看到的是一个空库而 Hive 里明明有表——这个现象我第一次遇到时排查了半小时。spark-sql -e show databases能列出 Hive 里的库说明集成成功。参数上spark.sql.warehouse.dir要和 Hive 的hive.metastore.warehouse.dir保持一致否则建表时数据目录会对不上。2.3 四类原始数据的接入与建表一卡通消费、图书借阅、门禁日志三类数据通常以文本或 CSV 形式导出先建外部表指向原始文件再做清洗。建表时全部用STRING类型接原始字段清洗阶段再转型这样能避免导入时因脏数据直接失败。-- 一卡通消费记录原始表 CREATE EXTERNAL TABLE ods_card_consume ( card_id STRING COMMENT 卡号, stu_id STRING COMMENT 学号, consume_time STRING COMMENT 消费时间, amount STRING COMMENT 金额, place STRING COMMENT 消费地点 ) ROW FORMAT DELIMITED FIELDS TERMINATED BY , STORED AS TEXTFILE LOCATION /user/hive/warehouse/ods/card_consume; -- 图书借阅记录原始表 CREATE EXTERNAL TABLE ods_book_borrow ( stu_id STRING, book_id STRING, borrow_time STRING, return_time STRING ) ROW FORMAT DELIMITED FIELDS TERMINATED BY , STORED AS TEXTFILE LOCATION /user/hive/warehouse/ods/book_borrow; -- 图书馆门禁日志原始表 CREATE EXTERNAL TABLE ods_gate_log ( stu_id STRING, gate_id STRING, pass_time STRING ) ROW FORMAT DELIMITED FIELDS TERMINATED BY , STORED AS TEXTFILE LOCATION /user/hive/warehouse/ods/gate_log;三张表都用外部表删表不删数据方便反复调试。字段全 STRING 是刻意的真实导出数据里金额可能带货币符号、时间可能混着2024/09/01和2024-09-01两种格式直接定义成 DECIMAL 或 TIMESTAMP 会在加载时报错。清洗层再统一转型问题定位也更清晰。3. 用 Scala 做多维度清洗与特征工程3.1 清洗的四个维度去重、补空、时间归一、异常剔除清洗不是一把梭要按维度拆开做。消费记录的核心问题是重复刷卡和金额异常借阅记录的问题是未归还return_time 为空门禁日志的问题是同一人短时间多次进出。我一般把清洗拆成四步每步一个 Scala 函数方便单独测试。import org.apache.spark.sql.SparkSession import org.apache.spark.sql.functions._ import org.apache.spark.sql.expressions.Window val spark SparkSession.builder() .appName(CampusDataClean) .enableHiveSupport() .getOrCreate() // 1. 消费记录清洗去重 金额转 Double 时间归一 val consumeRaw spark.table(ods_card_consume) val consumeClean consumeRaw .withColumn(amount_d, regexp_replace(col(amount), [^0-9.], ).cast(double)) .withColumn(consume_ts, to_timestamp( regexp_replace(col(consume_time), /, -), yyyy-MM-dd HH:mm:ss)) .filter(col(amount_d) 0 col(amount_d) 500) // 剔除异常金额 .dropDuplicates(card_id, consume_ts, amount_d) // 同卡同时间同金额视为重复 .filter(col(consume_ts).isNotNull) consumeClean.write.mode(overwrite).saveAsTable(dwd_card_consume)regexp_replace先把金额里的非数字字符清掉再 cast比直接 cast 稳。金额上限设 500 是经验值校园单笔消费超过 500 的极少大概率是数据错误或测试数据。去重用的三字段组合是因为同一张卡在同一秒刷两次相同金额基本可以判定为重复上报。to_timestamp前先替换斜杠兼容两种日期格式。3.2 借阅与门禁数据的对齐处理借阅记录里未归还的书 return_time 为空不能直接丢要标记成「在借」状态保留。门禁日志要做的是按人按天聚合算出每天首次进馆和末次离馆时间这是后面判断生活规律的关键特征。// 借阅记录空归还时间标记为在借 val borrowClean spark.table(ods_book_borrow) .withColumn(borrow_ts, to_timestamp(col(borrow_time), yyyy-MM-dd HH:mm:ss)) .withColumn(return_ts, to_timestamp(col(return_time), yyyy-MM-dd HH:mm:ss)) .withColumn(is_returned, when(col(return_ts).isNull, 0).otherwise(1)) .filter(col(borrow_ts).isNotNull) // 门禁日志按人按天算首次进馆、末次离馆 val gateDaily spark.table(ods_gate_log) .withColumn(pass_ts, to_timestamp(col(pass_time), yyyy-MM-dd HH:mm:ss)) .filter(col(pass_ts).isNotNull) .withColumn(dt, to_date(col(pass_ts))) .groupBy(stu_id, dt) .agg( min(pass_ts).as(first_in), max(pass_ts).as(last_out), count(*).as(pass_cnt) )门禁聚合用groupBy agg一次算出三个指标比多次窗口函数快。pass_cnt是当天进出次数次数异常高比如超过 20 次的可能是设备抖动后面特征工程里可以过滤。借阅的is_returned标记保留下来在特征里可以算「在借比例」反映学生的阅读活跃度。3.3 构建学生维度的宽表特征清洗完的三张表要按学号拼成一张宽表每个学生一行列是各类行为特征。这是喂给 KMeans 的最终输入。// 消费特征月均消费、消费频次、单笔均值 val consumeFeat consumeClean .groupBy(stu_id) .agg( sum(amount_d).as(total_amount), count(*).as(consume_cnt), avg(amount_d).as(avg_amount), countDistinct(to_date(col(consume_ts))).as(consume_days) ) // 借阅特征借阅总数、在借比例 val borrowFeat borrowClean .groupBy(stu_id) .agg( count(*).as(borrow_cnt), avg(is_returned).as(return_rate) ) // 门禁特征平均进馆次数、平均在馆时长小时 val gateFeat gateDaily .withColumn(stay_hours, (unix_timestamp(last_out) - unix_timestamp(first_in)) / 3600.0) .groupBy(stu_id) .agg( avg(pass_cnt).as(avg_pass_cnt), avg(stay_hours).as(avg_stay_hours) ) // 三表关联成宽表 val studentWide consumeFeat .join(borrowFeat, Seq(stu_id), left) .join(gateFeat, Seq(stu_id), left) .na.fill(0.0) studentWide.write.mode(overwrite).saveAsTable(dws_student_feature)关联用left join以消费表为主因为不是每个学生都借书或频繁进馆缺失的用 0 填充。na.fill(0.0)放在最后统一处理避免每个 join 后都填一次。宽表落成 Hive 表KMeans 直接从这张表读特征。注意stay_hours可能为负——如果门禁数据里 last_out 早于 first_in说明当天数据有跨天问题这个在避坑章节细说。4. KMeans 聚类的实现与参数调优4.1 特征向量组装与标准化KMeans 是基于距离的算法特征量纲不统一会直接毁掉聚类结果。消费金额动辄几百上千借阅次数只有个位数不标准化的话距离几乎全由消费金额主导其他维度等于没参与。import org.apache.spark.ml.feature.VectorAssembler import org.apache.spark.ml.feature.StandardScaler val featureCols Array( total_amount, consume_cnt, avg_amount, consume_days, borrow_cnt, return_rate, avg_pass_cnt, avg_stay_hours ) val assembler new VectorAssembler() .setInputCols(featureCols) .setOutputCol(raw_features) val scaler new StandardScaler() .setInputCol(raw_features) .setOutputCol(features) .setWithMean(true) .setWithStd(true) val assembled assembler.transform(spark.table(dws_student_feature)) val scaled scaler.fit(assembled).transform(assembled)setWithMean(true)做零均值化setWithStd(true)做方差归一两个都开才是标准正态标准化。如果特征里有明显的长尾分布比如消费金额标准化前可以先做 log 变换否则少数极端值会把均值拉偏。这一步的产物scaled里features列就是 KMeans 的输入。4.2 K 值选择肘部法加轮廓系数双验证K 值不能拍脑袋定。我一般同时跑肘部法和轮廓系数两个指标指向的 K 值一致才采信。肘部法看的是簇内平方和WSSSE随 K 下降的拐点轮廓系数看的是簇间分离度。import org.apache.spark.ml.clustering.KMeans import org.apache.spark.ml.evaluation.ClusteringEvaluator val evaluator new ClusteringEvaluator() .setFeaturesCol(features) .setPredictionCol(prediction) val results (2 to 8).map { k val kmeans new KMeans() .setK(k) .setSeed(42L) .setMaxIter(50) .setTol(1e-4) .setFeaturesCol(features) val model kmeans.fit(scaled) val wssse model.summary.trainingCost val pred model.transform(scaled) val silhouette evaluator.evaluate(pred) (k, wssse, silhouette) } results.foreach { case (k, w, s) println(fK$k WSSSE$w%.2f Silhouette$s%.4f) }setSeed(42L)固定随机种子保证每次跑结果可复现不然调参时 K 值没变结果却变了会误判。setMaxIter(50)对万级样本足够收敛setTol(1e-4)是中心点移动的收敛阈值。WSSSE 随 K 单调下降要找下降变缓的拐点轮廓系数在 -1 到 1 之间越接近 1 越好。校园数据一般 K3 到 5 比较合理对应「高消费活跃」「中等规律」「低消费宅」这类分群。4.3 聚类结果解读与业务映射模型跑完拿到 prediction 列要把它映射回业务含义才有价值。做法是算每个簇在各特征上的均值对比着看。val finalModel new KMeans().setK(4).setSeed(42L).fit(scaled) val clustered finalModel.transform(scaled) clustered.groupBy(prediction) .agg( avg(total_amount).as(avg_total), avg(consume_cnt).as(avg_cnt), avg(borrow_cnt).as(avg_borrow), avg(avg_stay_hours).as(avg_stay), count(*).as(stu_num) ) .orderBy(prediction) .show()看这张表某簇 avg_total 高、avg_cnt 高、avg_stay 也高就是消费活跃且常泡图书馆的群体某簇各项都低可能是低消费、少进馆的群体。簇编号本身没有意义靠特征均值来命名。stu_num看各簇人数是否均衡如果某一簇只有几个人说明 K 值偏大或存在离群点要回头检查特征。5. 避坑与排查一卡通数据清洗最容易翻车的五个点5.1 时间格式混用导致 to_timestamp 返回 null现象清洗后大量记录的consume_ts为 null过滤后数据量骤减。 原因原始导出数据里时间格式不统一同一列里混着2024-09-01 12:30:00和2024/09/01 12:30单一格式串解析不了。 解决先用regexp_replace把斜杠统一成横杠再用to_timestamp解析解析后统计 null 比例超过 5% 就说明格式还有漏网的把 distinct 的时间样本打出来看。5.2 门禁跨天记录算出负的在馆时长现象stay_hours出现负值聚类时这些学生被分到异常簇。 原因学生晚上进馆、次日凌晨离馆按自然日 groupBy 后first_in 是当天 23:00last_out 是当天更早的时间因为次日记录被分到另一天相减为负。 解决门禁聚合不按自然日切改成按「进馆事件」配对或者对负值直接置 0 并单独标记。我一般加一个when(stay_hours 0, 0)兜底同时统计负值比例比例高说明数据采集本身有问题。5.3 小文件过多拖垮 Hive 查询现象清洗后的 dwd 表目录下几千个小文件后续 Spark 读表时 task 数爆炸一个简单查询跑几分钟。 原因Spark 写入 Hive 时默认按分区数输出文件每个 task 一个文件数据量不大但 task 多就产生大量小文件。 解决写入前用repartition或coalesce控制文件数按数据量估算单文件 128MB 左右合适。或者在写入后跑 Hive 的concatenate合并。这个坑在迭代清洗逻辑时特别容易积累建议每次 overwrite 前都 coalesce 一下。5.4 特征标准化用了 fit 之外的 transform现象训练集和测试集或不同批次数据的聚类结果对不上同一学生在两次跑批里分到不同簇。 原因StandardScaler 的fit会计算均值和方差如果每次对新数据单独 fit标准化基准就变了距离计算自然不一致。 解决把 scaler 模型持久化新数据一律用同一个 scaler 做 transform。生产环境里 scaler 和 KMeans 模型要一起保存用model.save落盘下次直接 load。5.5 KMeans 对离群点敏感导致簇被拉偏现象某一簇只有两三个学生且这些学生消费金额极高或门禁次数极多。 原因KMeans 用均值做中心点离群点会把中心拉向自己形成孤立簇。 解决聚类前先做离群点检测用 IQR 或分位数把极端值截断winsorize或者对长尾特征做 log 变换。校园场景里消费金额超过 99 分位数的记录先截断到分位值再进模型。6. 让聚类结果真正可用的两个进阶技巧第一个技巧是给 KMeans 加业务约束。纯 KMeans 分出来的簇是数学最优但不一定业务可解释。我的做法是先跑一遍无约束聚类看各簇特征均值如果出现「消费高但借阅为零」这种矛盾簇就说明特征里有噪声维度在干扰。这时候可以手动降维把相关性高的特征比如 total_amount 和 avg_amount只保留一个或者用 PCA 先降到 4 到 5 维再聚类。PCA 在 Spark MLlib 里是PCA算子setK设成保留的主成分数一般保留累计方差贡献 85% 以上。第二个技巧是结果稳定性验证。KMeans 对初始中心敏感单次结果可能是运气。我习惯用不同的 seed 跑 5 次看同一批学生的簇归属一致率。一致率低于 80% 说明 K 值或特征有问题要回头调。代码上就是循环换 seed把每次的 prediction 收集起来算众数占比。val stability (1 to 5).map { seed val m new KMeans().setK(4).setSeed(seed.toLong).fit(scaled) m.transform(scaled).select(stu_id, prediction) } // 按 stu_id 对齐后算每个学生 5 次预测的众数占比 val joined stability.reduce((a, b) a.join(b, Seq(stu_id), inner)) // 后续用 UDF 或 groupBy 统计一致率这段代码把 5 次结果按 stu_id 拼起来每个学生有 5 个 prediction 值众数占比就是稳定度。稳定度高的学生可以放心用聚类标签做后续运营稳定度低的单独标记出来人工复核。最后说个我自己的习惯每次清洗逻辑改完不要直接 overwrite 生产表先写到临时表用except对比新旧表的差异行数确认改动符合预期再切换。一卡通数据涉及学生隐私和后续分析一次误覆盖可能让几天的清洗白做。这个后悔药我吃过一次之后所有清洗脚本都加了临时表对比这一步。希望帮到你。本文还有配套的精品资源点击获取