简介一份基于Spark的信用卡评分数据分析课程设计资源面向大数据、Python及Spark的入门学习者也适合需要完成类似课程设计、毕业设计或实践项目的学生参考。资源以和鲸社区公开的信用卡评分模型构建数据为基础包含数据预处理、数据分析与可视化等环节的完整Python脚本覆盖数据清洗、特征观察、结果图表生成等常见任务延展了年龄、收入、不动产贷款、逾期情况、家庭人口等维度分析。压缩包共22个文件以py源码、html可视化页面、csv数据文件和doc课程设计报告为主另有少量工程配置文件与说明文件整体约4.91MB便于按用途快速定位当前已有3781人学习下载。通过代码与报告的相互对照可以理解Spark处理结构化数据的基本流程也能直接复用样式与逻辑减少从零搭建项目的成本。配套的课程设计报告对项目背景、实现思路和结果分析有较完整说明适合把该案例作为学习模板或二次开发基础。1. 基于Spark的信用卡评分数据分析先搞清楚要解决什么问题很多人一拿到信用卡评分这类项目第一反应是打开 Pandas 开始跑groupby但数据量一旦到千万级交易流水、上百万用户单机内存立刻见底跑一个聚合要等十几分钟更别说后面还要叠特征工程和模型训练。基于Spark的信用卡评分数据分析就是把数据清洗、特征加工、模型训练这三段重活全部挪到分布式集群上做让数据量和计算时长不再是瓶颈。这篇文章面向刚接触 Spark 的开发者和风控从业者帮你把一套能落地的评分数据分析路径走通先建管道再写 ETL然后训练模型最后避开我在实际项目中踩过的那些坑。2. 为什么选 Spark 做评分数据分析分布式计算的三个关键选型理由2.1 数据规模决定工具链从单机 Pandas 到 Spark DataFrame 的迁移判断信用卡评分数据分析里最常被低估的是数据量。一张信用卡一个月的交易流水就可能上万条一个中等规模的银行或消金公司用户数在百万级流水表轻松破亿。Pandas 处理一亿行数据光read_csv就要吃掉十几 G 内存groupby之后的内存峰值更没法控制。我见过有同事用 Pandas 跑一个月的用户聚合最后进程被 OOM killer 干掉代码白跑。判断要不要上 Spark就看三条数据量是不是超过单机内存的一半、计算逻辑里有没有多步聚合和连接、后续模型训练要不要反复复用同一份特征。三条里命中两条就值得迁到 Spark。Spark 的 DataFrame API 和 Pandas 很像select、filter、groupBy、join的语义几乎一致迁移成本比想象中低但有个关键差异必须先建立Pandas 是 eager 执行Spark 是 lazy 执行。你写df.filter(...)的时候Spark 并不会立刻算只有遇到count()、collect()、write这类 action 才会真正触发计算。from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(card_score_etl) \ .config(spark.sql.shuffle.partitions, 400) \ .config(spark.sql.adaptive.enabled, true) \ .getOrCreate() # 读取交易流水这段只是构建逻辑计划不会真正读数据 txn_df spark.read.parquet(/data/raw/txn_log) filtered txn_df.filter(txn_df.txn_amt 0)这段代码里的.config(spark.sql.shuffle.partitions, 400)值得说一句。shuffle 分区数直接决定后续 groupBy、join 的并发度。默认值 200 在几亿行数据上不够用单分区处理几千万行会导致个别 task 跑得特别慢。设 400 只是起步值最终要根据 executor 核心数来定一般是 executor 总核心数的 2 到 3 倍。.config(spark.sql.adaptive.enabled, true)开启 AQE 之后Spark 会在运行时根据 shuffle 文件大小动态合并分区这个我建议无论什么项目都打开它能自动缓解很多分区设得不合理的状况。2.2 DataFrame 与 RDD 的取舍评分数据管道为什么用 API 而不是底层算子信用卡评分数据管道里有人在网上推荐用 RDD 做所有计算理由是灵活。但我实际做下来除非你要操作二进制数据或者写非常底层的自定义分区逻辑否则不要碰 RDD。DataFrame API 背后有 Catalyst 优化器它会自动做谓词下推、列剪枝、常量折叠这些优化在 RDD 层面全部要自己手写。举个例子如果你只需要user_id和txn_amt两列用 DataFrame 写select(user_id, txn_amt)Spark 读 parquet 文件时只反序列化这两列用 RDD 的话整行数据都会被读进来I/O 开销差出好几倍。from pyspark.sql.functions import col, to_date, sum as spark_sum # 读取日期分区直接过滤分区字段而不是扫描全表 txn_df spark.read.parquet(/data/raw/txn_log) \ .filter(to_date(col(txn_time)) 2024-01-01) # 选列裁剪 用户粒度聚合 user_daily_amt txn_df.select(user_id, txn_amt) \ .groupBy(user_id) \ .agg(spark_sum(txn_amt).alias(total_amt_30d))这里的.filter(to_date(col(txn_time)) 2024-01-01)有个性能细节。如果原始表是按天分区的目录结构/data/raw/txn_log/dt2024-01-01/应该直接写filter(col(dt) 2024-01-01)这样 Spark 只会读取匹配的分区目录如果像上面这样对txn_time做to_date再比较Spark 无法做分区裁剪只能全表扫描后再过滤数据量大的时候作业时间会翻好几倍。这是个非常容易踩的坑后面避坑章节还会细说。选 DataFrame 还有一个实际好处写出来的代码更接近 SQL 思维团队里做数据分析出身的人基本不需要额外学习就能 review 你的逻辑。我们当时的做法是让 Python 开发者统一用 PySpark DataFrame APISQL 开发者直接用 Spark SQL 注册临时表两套逻辑共用同一套底层执行引擎协作效率明显比各写各的高。3. 信用卡评分数据 ETL 实战从原始流水到用户特征宽表3.1 原始数据清洗空值、重复交易与异常金额的三层过滤评分数据里的脏数据比想象中严重。我接手过一个数据源同一笔消费被上游系统重推了三次如果不做去重用户消费金额会被高估评分模型的收入类特征直接失真。清洗逻辑一般分三层先按主键去重再处理空值和默认值最后过滤明显异常的业务数据。主键通常选(user_id, txn_id, txn_time)组合txn_id单独做主键不保险因为不同系统对同一笔交易生成的 ID 可能不同。from pyspark.sql.functions import col, count, when, isnan # 第一层组合主键去重保留最早一笔 deduped txn_df.dropDuplicates([user_id, txn_id, txn_time]) # 第二层空值策略金额空的按 0 处理商户字段空的填 UNKNOWN cleaned deduped \ .withColumn(txn_amt, when(col(txn_amt).isNull(), 0).otherwise(col(txn_amt))) \ .withColumn(merchant_id, when(col(merchant_id).isNull(), UNKNOWN).otherwise(col(merchant_id))) # 第三层过滤异常金额单笔超过 100 万直接剔除 valid cleaned.filter(col(txn_amt).between(0, 1000000))dropDuplicates([user_id, txn_id, txn_time])的语义是按这三列的组合去重保留第一条。这里有个细节Spark 的去重是保留任意一条不保证是最早的一条。如果你像上面代码那样希望保留最早一笔要先按txn_time排序但全局排序在亿级数据上很贵。我一般会改用窗口函数row_number配合分区排序去解决from pyspark.sql import Window from pyspark.sql.functions import row_number window_spec Window.partitionBy(user_id, txn_id) \ .orderBy(col(txn_time).asc()) deduped txn_df.withColumn(rn, row_number().over(window_spec)) \ .filter(col(rn) 1) \ .drop(rn)用窗口函数做去重partitionBy指定去重的分组键orderBy决定保留哪一条。这里asc表示保留最早一条。窗口去重比dropDuplicates慢但在需要确定性结果的场景下值得。还有一个容易被忽视的点when(col(txn_amt).isNull(), 0)对 Spark 的 NaN 不生效NaN 要用isnan()判断。如果上游数据里有 NaN 混进来空值处理会把 NaN 原样保留后面聚合出来的金额就成了 NaN模型训练直接报错。保险起见我会把两个判断合并when(isnan(col(txn_amt)) | col(txn_amt).isNull(), 0)。3.2 时间窗口聚合特征为什么选择 30 天滑动窗口而不是固定月度信用卡评分里最常用的特征类别就是消费行为汇总比如近 30 天消费总额、交易笔数、最大单笔金额、平均每笔金额。固定按月聚合有个问题每月的天数和用户账单日不对齐导致提取的特征分布有月度波动。比如 2 月只有 28 天和 3 月的 31 天相比消费笔数天然偏少模型学到的是月份效应而不是用户行为。所以我做评分特征一律用相对当前时间的滑动窗口最常用的是往前推 30 天、90 天和 180 天三檔。from pyspark.sql.functions import current_date, datediff, lit # 先给每笔交易打上相对天数标签 txn_with_days valid.withColumn( days_ago, datediff(current_date(), col(txn_time).cast(date)) ) # 30 天窗口聚合 feature_30d txn_with_days \ .filter(col(days_ago) 30) \ .groupBy(user_id) \ .agg( spark_sum(txn_amt).alias(amt_30d), count(txn_id).alias(cnt_30d), max(txn_amt).alias(max_amt_30d) )用datediff(current_date(), col(txn_time))计算每条记录离当前日期的天数然后filter(days_ago 30)直接框定窗口比用between日期区间更直观而且在分区裁剪良好的情况下性能差别不大。但有个前提current_date()在同一个 Spark 作业里是固定值不会因为跑的时间不同而改变这保证了特征的确定性特征。当时我们做这件事踩过一个坑如果用date_sub(current_date(), 30)去算窗口起点Spark 会把日期常量下推到 scan但一旦current_date的语义理解错下次重跑作业拿到的窗口边界可能和上次不同。建议在作业开头把current_date()的值取出来存成变量统一用这个变量做窗口计算。实际项目里我会把 30 天、90 天、180 天的聚合分别写成一个函数再 three 张表 JOIN 到同一张用户宽表上。JOIN 的时候注意用broadcast把小表广播出去避免每次都 shuffle。4. Spark MLlib 训练评分模型逻辑回归与阈值调参完整流程4.1 特征向量化与归一化注意 StringIndexer 的排序陷阱特征宽表建好之后接下来用 Spark MLlib 训练评分模型。信用卡评分领域最常用的还是逻辑回归原因有三个可解释性强、训练快、对稀疏特征友好。但逻辑回归对数值尺度敏感特征里有消费金额这种几十万的数值也有交易笔数这种几十的数值如果不做归一化梯度下降收敛会非常慢甚至不收敛。MLlib 里用VectorAssembler把多列特征拼成一个向量然后接StandardScaler做归一化。from pyspark.ml.feature import VectorAssembler, StandardScaler from pyspark.ml.classification import LogisticRegression # 数值特征统一做 z-score 归一化 num_features [amt_30d, cnt_30d, max_amt_30d, amt_90d, cnt_90d] assembler VectorAssembler(inputColsnum_features, outputColfeatures_raw) scaler StandardScaler( inputColfeatures_raw, outputColfeatures, withStdTrue, withMeanTrue ) # 逻辑回归设定正样本权重缓解样本不平衡 lr LogisticRegression( featuresColfeatures, labelColis_default, maxIter50, regParam0.01, weightColweight )StandardScaler的withMeanTrue会做中心化让每个特征均值变 0 方差变 1。但这里有个注意点withMeanTrue在稀疏向量上会触发稠密化如果特征维度特别高比如几百个整个向量会变成稠密向量内存占用暴涨。碰到高维特征的情况我建议只做withStdTrue或者改用MinMaxScaler。另外LogisticRegression里的weightCol是处理样本不平衡的关键。信用卡违约样本通常只有 2% 到 5%如果不加权模型会全部预测为正常用户AUC 看起来还行但精确率几乎为 0。权重一般这么算正样本权重 总样本数 / (2 × 正样本数)负样本权重 总样本数 / (2 × 负样本数)。4.2 训练验证切分与模型评估AUC 不够还要看 KS 和分数分布模型训练完不能只看 AUC。风控场景里评分卡最终要把概率映射成整数分数比如 300 到 800 分然后设定一个审批阈值。AUC 衡量的是排序能力但排序能力好不代表你在某个阈值下的表现好。业界常用的还有 KS 统计量衡量好样本和坏样本分数分布的分离程度。KS 大于 0.3 勉强可用大于 0.4 算不错。实际验证逻辑是把训练好的模型在测试集上transform拿到每个用户的违约概率用BinaryClassificationEvaluator算 AUC同时手动算 KS。from pyspark.ml.evaluation import BinaryClassificationEvaluator # 划分训练与测试 train_df, test_df features_df.randomSplit([0.8, 0.2], seed42) model lr.fit(train_df) predictions model.transform(test_df) evaluator BinaryClassificationEvaluator( labelColis_default, metricNameareaUnderROC ) auc evaluator.evaluate(predictions) print(ftest AUC {auc:.4f})randomSplit([0.8, 0.2], seed42)里的seed一定要固定否则每次跑出来的切分不一样模型对比就没法做了。我习惯把训练集、验证集、测试集按 [0.7, 0.15, 0.15] 切验证集用来调超参数测试集只做最终评估。还有一个经验测试集不要反复用调参时只看验证集最后跑一次测试集记录结果否则容易过拟合测试集。在信用卡评分里这个过拟合不是模型对数据的过拟合而是你自己在拿测试集反复调阈值导致换一批新数据效果暴跌。我们内部管这叫「用测试集调参的玄学」后来公司强制要求测试集只能跑一次效果稳定多了。模型训练完下一步就是把概率映射成分数。标准做法是用probability列中正类概率 p按公式score 600 round((log(p/(1-p)) - offset) * factor)换算offset和factor是根据业务基准分和刻度设定的常量。这个映射逻辑一般不用 Spark 实现在模型落库时用 UDF 处理即可。5. 避坑Spark 评分作业中频繁出现的五个问题与排查思路5.1 数据倾斜导致单个 task 跑几个小时现象作业进程整体进度卡在 99%只有一个 task 迟迟不结束其他 task 早就跑完了。原因信用卡数据里头部用户的交易量极多比如某些高端卡用户一个月几千笔交易groupBy(user_id)时这些用户的数据全部分到同一个分区单 task 数据量是其他 task 的几百倍。解决先确认是不是数据倾斜在 Spark UI 里看 stage 的 task 耗时分布如果最长耗时和平均耗时差一个数量级基本就是倾斜。处理方法是给 user_id 加盐把大 key 拆散到多个分区聚合完再去掉盐前缀。from pyspark.sql.functions import concat, lit, rand # 给 user_id 加 10 个随机盐值分散热点 salted valid.withColumn( salt_key, concat(col(user_id), lit(_), (rand() * 10).cast(int)) ) # 先按加盐 key 聚合再去盐后二次聚合 aggregated salted.groupBy(salt_key) \ .agg(spark_sum(txn_amt).alias(amt)) \ .groupBy(user_id) \ .agg(spark_sum(amt).alias(amt_total))(rand() * 10).cast(int)生成 0 到 9 的整数盐值乘 10 表示把热点 key 拆成 10 份。这个数字不是随便定的如果热点 key 的数据量占到总数的一半盐值至少要设成ceil(最大分区数据量 / 平均数据量)。加盐能解决问题但会让 Job 多跑一个 stage所以只针对确认倾斜的 key 做不要全表加盐。5.2 分区字段被函数包裹导致分区裁剪失效现象跑了同样一段 SQL在某个数据源上只要 10 分钟换了个表却要 1 小时看 Spark UI 发现读取的文件数量完全没减少。原因分区字段在where条件里被包了函数。比如分区字段是dt有人写成where substr(dt, 1, 7) 2024-01Spark 无法把这个条件下推成分区过滤只能全表扫描。解决不要在分区字段上套函数直接用原始字段比较。这也是我在前面 ETL 部分强调过的点但实际项目里反复出现因为数据工程师习惯用substr和to_date加工字段。5.3 Spark 作业频繁 OOM 但加了内存参数也没用现象executor 报java.lang.OutOfMemoryErrorspark.executor.memory已经调到 8G 还是崩。原因大多数情况下不是执行内存不够而是 shuffle 的中间文件或者广播变量失控。spark.sql.autoBroadcastJoinThreshold默认 10MB如果有一张特征表刚好超过阈值没被广播JOIN 就会触发全量 shuffle大表 OOM。解决先用explain()看执行计划确认 JOIN 是SortMergeJoin还是BroadcastHashJoin再显式给小表加broadcast()提示。from pyspark.sql.functions import broadcast # 显式广播用户维表避免 shuffle joined feature_df.join(broadcast(user_dim_df), user_id, left)broadcast(user_dim_df)是手动告诉 Spark 把 user_dim_df 复制到每个 executor 上。适合广播的表一般要求单表大小在几百 MB 以内太大会导致 driver 端发送数据就 OOM。另外还有个小技巧如果设置spark.sql.adaptive.enabledtrue且spark.sql.adaptive.coalescePartitions.enabledtrueSpark 会在 shuffle 后自动合并小分区减少后续读取压力。5.4 模型训练时 StringIndexer 对测试集报错未命中标签现象模型在训练集上一切正常做预测时StringIndexer抛unseen label异常。原因StringIndexer默认策略是会处理未知标签但如果你用了setHandleInvalid(keep)却忘了在预测流水线上保持同一份映射测试集里出现的新类别就会翻车。解决把特征工程和模型封装成一个Pipeline这样拟合和预测用的是同一个转换逻辑。from pyspark.ml import Pipeline pipeline Pipeline(stages[indexer, assembler, scaler, lr]) pipeline_model pipeline.fit(train_df) predictions pipeline_model.transform(test_df)Pipeline 的价值就是保证训练和预测两阶段走完全相同的转换逻辑。信用卡评分里新用户、新商户的类别随时会出现如果不用 Pipeline 而是手动保存每个 Stage 的模型漏掉一个就会在线下正常、线上报错。5.5 重跑作业结果对不上窗口边界和采样扰动现象同一个作业昨天跑和今天跑模型指标差 0.02特征分布也有细微差异。原因两条一是时间窗口边界用的是动态current_date()昨天跑的日期窗口和今天的不一致二是randomSplit没固定seed每次切分样本不同。解决在作业代码开头把日期基准和随机种子用参数传入这样任何一次重跑都能回放。这也是生产环境里要求作业可重复的基本功。6. 上线后的持续验证用 Spark 定期重算 PSI 与分数分布漂移模型上线只是开始信用卡评分模型的失效速度比预期快得多。消费习惯、宏观环境、用户结构都在变上个月还稳定的评分分布下个月可能整体偏移。我习惯在模型上线后每周用 Spark 重算一次 PSIPopulation Stability Index监控当前评分分布与建模时基准分布的偏离程度。计算逻辑不复杂把当前分数分箱算出每个箱的占比和基准箱占比做对比求和。from pyspark.sql.functions import col, when, lit import math # 假设基准分布已经按分数区间算好并存入字典 benchmark_bins { 0-300: 0.05, 300-400: 0.10, 400-500: 0.20, 500-600: 0.35, 600-700: 0.20, 700-800: 0.10 } # 当前模型打分结果按同样区间分箱 current_bins predictions.withColumn( score_bin, when(col(score) 300, 0-300) .when(col(score) 400, 300-400) .when(col(score) 500, 400-500) .when(col(score) 600, 500-600) .when(col(score) 700, 600-700) .otherwise(700-800) ) bin_counts current_bins.groupBy(score_bin).count() total current_bins.count() psi_value 0.0 for row in bin_counts.collect(): bin_name row[score_bin] actual_ratio row[count] / total expected_ratio benchmark_bins[bin_name] psi_value (actual_ratio - expected_ratio) * math.log(actual_ratio / expected_ratio) print(fPSI {psi_value:.4f})PSI 的判定经验值小于 0.1 表示分布稳定0.1 到 0.25 之间需要关注超过 0.25 说明分布发生了显著漂移模型要重新训练了。上面的when链写的是分数区间映射注意区间边界要闭合否则会有用户落不到任何箱里。collect()会把结果拉到 driver如果分箱数量少一般不超过 20 箱没问题但如果要对全量用户collect()那就是灾难。这里只拉 7 个箱的聚合结果数据量很小。如果再细致一点我会把 PSI 的监控写成一个独立的 Spark 作业入参是模型版本和日期输出是一张监控表包含各分数箱的占比和 PSI 值。一旦 PSI 超过 0.25自动触发告警。这个作业我坚持做了三个月还真抓到一个线上事故某次渠道策略调整导致一批低分用户集中涌入整体分数分布肉眼可见地左移PSI 飙到 0.31模型审批阈值没动坏账率直接上升。从那以后我每个评分模型上线的第一天就会先把 PSI 监控作业跑起来基线分布、分箱逻辑、告警阈值全部固化在代码里每周定时调度。这个习惯帮我避开了至少两次模型悄悄失效的风险。希望这篇基于 Spark 的信用卡评分数据分析实战笔记也能帮你在自己的数据管道上少走几步弯路。本文还有配套的精品资源点击获取