简介面向金融大数据与风控开发者的完整源码包基于Hadoop与Spark构建信贷风险控制系统覆盖数据摄入、预处理、模型训练、风险评估和可视化等完整链路适合正在研究金融风控场景的Hadoop与Spark开发者参考借鉴。压缩包共69个文件以36个Java、8个Scala源码文件为主体另有12个XML配置、5个Properties配置以及SQL、JSON、JS等辅助文件整体仅72KB目录结构清晰便于快速定位批处理、实时流计算和Web展示相关模块。项目还涉及信用评估、欺诈检测、还款能力分析等典型场景数据清洗、特征工程、模型训练与结果展示均有对应代码组织。目前已有1219人学习下载。通过阅读源码可以了解HDFS与Spark Streaming的配合方式、逻辑回归与随机森林等违约预测模型的工程实现也能看到多源数据集成和敏感数据保护的基础落地思路适合边看边改、快速上手。1. 基于 Hadoop、Spark 的信贷风控系统源码它解决的是“算不动”和“算不快”两件事信贷业务跑批最怕的不是模型不准是数据根本算不动。一条进件要关联近三年流水、多头借贷、历史逾期单机 Oracle 凌晨两点还卡在 join 上。基于 Hadoop、Spark 的大数据金融信贷风险控系统源码就是为这种场景准备的HDFS 扛原始流水Spark 做清洗、特征加工和评分把小时级批处理压到分钟级。它适合三类人想找完整大数据项目练手的新手、拿它做毕业设计基座的在校生、缺参考架构的团队。我按这类源码最常见的工程结构把跑通全流程的步骤、参数和踩坑顺序讲一遍。2. 风控系统的技术选型与工程目录为什么先搭 Hadoop 再跑 Spark2.1 HDFS、YARN、Spark 在风控系统里各管哪一段组件分工要先理清。HDFS 是存储层负责存进件申请表、资金流水、征信报文、历史还款记录文件以 JSON、Parquet 或 CSV 分目录落地。YARN 是资源调度层把 CPU 和内存切分成容器分配给计算任务。Spark 是计算引擎把原始数据处理成风控可用的特征宽表再跑模型批量打分。常见错误是把三件事混在一起比如拿 HDFS 当关系型数据库做随机查询或者让 Spark 任务直连业务库结果把 MySQL 压到连接数爆掉。我的习惯是业务库只做增量导出全量历史进 HDFSSpark 只跟 HDFS 打交道算完结果再写回业务库或结果表。组件在风控系统里的角色典型落点HDFS原始数据存储海量流水持久化/risk/input/loan、/risk/input/repayYARNCPU 与内存的资源池容器调度、队列配额控制Spark清洗、特征加工、模型评分Spark SQL 批处理、MLlib 模型训练权限组件行、列级权限与敏感字段脱敏HDFS 目录 ACL、Spark SQL 列级脱敏信贷数据和普通日志的敏感度不在一个级别数据血缘和行、列权限设计从第一天就要做。常见做法是 HDFS 目录按业务线切分并设置 ACL物理隔离不同团队敏感列脱敏放在 Spark SQL 层手机号和身份证用 regexp_replace 打码后再写结果表而不是等下游去处理。很多源码包把权限写死在代码里我建议你改成外部配置否则换环境就要改代码重新打包这是最典型的返工现场。2.2 从进件到风险分的五段数据流转数据流转大致分五段。第一段接入Kafka 或定时任务把业务库增量数据落到 HDFS按 dt 分区第二段清洗处理空值、去重、修正时间字段落到 ODS 层第三段特征按客户维度聚合出近三个月逾期次数、额度使用率、多头借贷笔数等落到特征宽表第四段评分Spark 加载特征宽表用训练好的模型批量打分输出风险分和拒绝理由码第五段回写结果写回业务库供审批系统调用同时留一份到 HDFS 做监控和复盘。每段之间要加检查点。特征层算完后先跑数据质量 SQL检查客户总数、空值率、金额字段的 min/max再往下游送。这一步能挡住绝大多数“模型分漂移但没人知道原因”的翻车现场。这个流程和农产品价格数据分析、网约车数据分析这类 Spark 数据分析案例套路一致区别只在业务口径更敏感、字段更长、跑批时间窗口更短。2.3 源码 zip 解压后的典型工程结构拿到源码 zip解压后常见工程结构是这样risk-control-system/ ├── conf/ │ ├── core-site.xml │ ├── hdfs-site.xml │ └── spark-defaults.conf ├── sql/ │ ├── ddl/ # 建表与分区语句 │ ├── feature.sql # 特征加工 SQL │ └── quality_check.sql # 数据质量检查 ├── jobs/ │ ├── ingest.py # 数据接入 │ ├── feature.py # 特征宽表 │ ├── train.py # 模型训练 │ └── score.py # 批量打分 ├── model/ │ └── lr_model/ # 训练产出的模型目录 ├── lib/ │ └── risk_utils.py # 公共工具类 └── README.md解压后第一件事不是看代码而是看 README 和 conf 目录。conf 里的 xml 决定这套代码在哪个环境跑端口、内存、副本数全在这里。jobs 目录是核心按文件名能看出执行顺序ingest 在最前feature 其次train 只在需要重训时跑score 是日常任务。lib 里的公共工具类通常包含日志封装和 HDFS 工具换环境时先改这里。代码逻辑上源码把 SQL 和 Python 分开SQL 只做数据加工Python 负责调 Spark、写 HDFS、读模型。这么拆的好处是特征口径变更时不用动调度代码数仓方向的同学也能直接改 SQL。接手这套源码时先跑一遍 sql/ddl 里的建表语句确认分区字段和文件路径对得上再动 Python能省下大量“任务失败在路径不存在”的血泪时间。3. 源码本地复现Hadoop 伪分布式搭建与 Spark 最小任务验证3.1 伪分布式环境的关键配置core-site、hdfs-site 与 Spark 内存参数复现这套源码最低成本的做法是伪分布式一台机器模拟一个 HDFS 集群和一个 YARN 集群Spark 跑在 YARN 上。新手最容易在配置参数上翻车我给你一组能跑通的最小配置。core-site.xml 里最关键的是 fs.defaultFS把默认文件系统指到 HDFSconfiguration property namefs.defaultFS/name valuehdfs://localhost:9000/value /property /configurationhdfs-site.xml 里必须把副本数改成 1。伪分布式只有一台 DataNode默认 3 会让 block 一直处于欠副本状态NameNode 动不动就进安全模式configuration property namedfs.replication/name value1/value /property property namedfs.namenode.name.dir/name value/opt/hadoop_data/namenode/value /property property namedfs.datanode.data.dir/name value/opt/hadoop_data/datanode/value /property /configuration内存配置是伪分布式最容易炸的地方yarn-site.xml 里要显式声明 NodeManager 可用内存property nameyarn.nodemanager.resource.memory-mb/name value4096/value /property property nameyarn.scheduler.maximum-allocation-mb/name value4096/value /propertyspark-defaults.conf 里我一般这样写spark.masteryarn spark.executor.memory1g spark.executor.cores1 spark.driver.memory1g spark.sql.shuffle.partitions20关键参数的含义要清楚spark.executor.memory 别超过 NodeManager 声明内存的一半否则 YARN 会直接杀掉 executorspark.sql.shuffle.partitions 默认是 200在伪分布式下会产生大量小文件我压到 20跑测试快一个量级。这套配置跑开发够用等真上了集群再按机器规格重调。也有人用 Docker 镜像拉一套现成的 Hadoop 环境能省安装时间但端口映射和容器内存限制反而比伪分布式更绕我建议第一次跑通用裸机伪分布式问题好排查。3.2 启动 HDFS 与 YARNjps 确认五类进程配置改完先格式化 NameNode 再启动。格式化只有第一次执行时安全之后别再跑否则元数据清空等于给已存的数据喂后悔药hdfs namenode -format start-dfs.sh start-yarn.sh jpsjps 输出里必须看到 NameNode、DataNode、SecondaryNameNode、ResourceManager、NodeManager 这五个进程。少一个都别往下走先看日志NameNode 日志在 /opt/hadoop/logs/hadoop-hadoop-namenode-*.log报错多半是格式化目录不一致或端口被占。确认进程齐全后用 spark-shell 验证 Spark 能正常提交任务spark-shell --master yarn --deploy-mode client进入 shell 后执行一句最简单的 SQLspark.sql(select 1 as a).show能打印出结果就说明 Spark 和 YARN 之间的链路通了后面跑源码里的 jobs 才有意义。这一步是伪分布式搭建之后最朴素的一次 Spark 实战验证也最适合拿来当环境是否可用的标准用例。到这步环境就算立住了。3.3 用 Spark 读取 HDFS 上的 JSON先跑通第一个风控小任务很多源码包的样例数据是 JSON 格式比如 loan_20250101.json 存当天进件。用 PySpark 读取并做一次聚合是最直接的验证方式from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(RiskLoanCheck) \ .getOrCreate() df spark.read.json(hdfs://localhost:9000/risk/input/loan_20250101.json) df.printSchema() df.groupBy(channel).count().show()这段代码做了三件事建立 Spark 会话、读取指定路径 JSON、按申请渠道统计数量。printSchema 能看出 Spark 推断出的字段类型这只是验证用的读法真实跑批时我会用 structType 显式指定 schema原因后面讲。路径里的 hdfs://localhost:9000 必须和 core-site.xml 里的 fs.defaultFS 一致否则会用本地文件系统跑直接报路径不存在。如果你要读多天文件把路径改成 loan_*.json 就能用通配符批量加载。跑通这个任务后把源码里 ingest.py 的输入路径改成你的样例数据路径先跑最小数据集不要一上来全量灌。伪分布式下单节点吞吐有限几百 MB 数据能验证逻辑就够了调优留给集群环境。4. 风控核心逻辑落地特征加工、模型训练与参数调优4.1 特征加工从流水明细到客户维度的宽表风控系统的建模主体是客户但原始数据是流水所以特征加工本质是一堆按客户维度做聚合的 Spark SQL。拿逾期特征举例常见做法是取近 90 天、近 180 天两个窗口分别统计逾期次数和平均逾期天数from pyspark.sql import functions as F repay spark.read.parquet(hdfs://localhost:9000/risk/ods/repay) loan spark.read.parquet(hdfs://localhost:9000/risk/ods/loan) feature_df loan.filter(F.col(due_dt) 2025-01-01) \ .join(repay.select(loan_id, overdue_days, repay_dt), loan_id, left) \ .groupBy(customer_id) \ .agg( F.sum(F.when(F.col(overdue_days) 0, 1).otherwise(0)).alias(ovd_cnt_90d), F.avg(overdue_days).alias(avg_ovd_days) )逻辑说明loan 表按到期日期过滤出近期进件跟 repay 表 left join按客户分组统计逾期次数和平均逾期天数。这里刻意没写还款日期的窗口过滤实际业务要在 join 前用 repay_dt 缩小数据量否则 join 的数据集太大。left join 保留所有进件客户避免有申请但还没还款记录的客户被丢掉。聚合函数里 F.sum(F.when(...)) 是条件计数比先 filter 再 groupBy 少一次 shuffle。avg 对 null 自动跳过符合逾期口径要求。如果字段里有极端值聚合前加一个 where overdue_days between 0 and 365 的约束防止脏数据把均值拉飞。特征宽表算完后直接 repartition(1) 写回 HDFS 会造成单文件过大本地验证可以这么干生产环境要按客户号 hash 分区到 20 个文件左右。4.2 评分模型训练MLlib 逻辑回归与关键参数特征宽表准备好以后进模型训练环节。源码包里最常见的模型是逻辑回归因为可解释性好信贷场景要跟监管解释拒绝原因。训练代码的关键部分from pyspark.ml import Pipeline from pyspark.ml.feature import VectorAssembler, StandardScaler from pyspark.ml.classification import LogisticRegression feature_cols [ovd_cnt_90d, avg_ovd_days, limit_usage, multi_loan_cnt] assembler VectorAssembler(inputColsfeature_cols, outputColraw_features) scaler StandardScaler(inputColraw_features, outputColfeatures, withStdTrue, withMeanTrue) lr LogisticRegression(featuresColfeatures, labelColis_default, regParam0.01, elasticNetParam0.5, maxIter50) pipeline Pipeline(stages[assembler, scaler, lr]) model pipeline.fit(train_df)先看 VectorAssembler它把四个数值特征拼成一个向量列。StandardScaler 把特征标准化到均值 0 方差 1对逻辑回归收敛速度影响明显尤其 limit_usage 这类量纲大的变量不标准化训练时 Loss 容易震荡。LogisticRegression 三个参数里regParam 是正则系数默认 0.01 在样本量小于十万时偏小我一般先给 0.05 试elasticNetParam 控制 L1/L2 混合比例0.5 表示各占一半maxIter 默认 100特征标准化后 50 轮通常就收敛。训练集和测试集切分有讲究。直接用 randomSplit 会随机打散客户导致同一个客户的数据同时出现在训练集和测试集里评估指标虚高。更稳的做法是按是否违约分层抽样保证正负样本比例在两份集合里一致否则测试集里正样本占比变了分数阈值就全偏了。信贷数据普遍存在逾期率低于 5% 的类别不平衡训练前先看 label 分布必要时对多数类做降采样。模型评估用 AUC 和 KSBinaryClassificationEvaluator 默认算 AUCKS 需要自己写累积分布逻辑。如果源码只打印 accuracy请只把它当辅助指标信贷场景里把所有人都预测为不逾期也有 95% 准确率这是最典型的假象。模型训练完要持久化方便 score.py 直接加载model.write().overwrite().save(hdfs://localhost:9000/risk/model/lr_model)这里 model 是 PipelineModel保存的是整条处理链路打分时自动包含标准化不用在 score.py 里再拼一遍特征处理。4.3 数据倾斜与内存调优repartition、广播变量与敏感列脱敏跑真实数据时最常见的性能杀手是数据倾斜。特征加工时按 customer_id 分组有些大客户流水量级比平均数高几个数量级一个 task 跑很久其他 task 空转。我一般分两步处理先在 join 前给关联键加盐打散热点聚合完再按真实键聚合回去或者简单点对大表做 repartition对维表做广播。广播变量在特征加工里很实用渠道信息、地区码这类维度表只有几千行用 broadcast 可以避免每个 executor 都拉一次from pyspark.sql.functions import broadcast result feature_df.join(broadcast(dim_channel), channel_code, left)默认的广播阈值由 spark.sql.autoBroadcastJoinThreshold 控制默认 10MB超过这个大小不会自动广播要手动在 join 时加 broadcast() 函数。动态分区裁剪和 AQEAdaptive Query Execution在 Spark 3 里默认开启如果源码包基于 Spark 2.4 老版本这些优化没有shuffle 分区要手动调。内存调优还有一个容易漏的点spark.executor.memoryOverhead 默认是 executor 内存的 10%跑大 shuffle 时实际占用可能超过申请值伪分布式下尤其明显。敏感字段脱敏必须放在这一层解决。写入结果表之前对手机号、身份证做打码result result.withColumn( mobile_masked, F.regexp_replace(mobile, (\\d{3})\\d{4}(\\d{4}), $1****$2) )这段正则保留手机号前三位和后四位中间打码。风控系统接入大数据行、列权限设计时脱敏一定要在 Spark 层做不能只靠数据库视图否则 Parquet 文件落在 HDFS 上能读文件的人直接看明文属于安全红线没有后悔药。5. 复现风控系统的排查手记5 个高频坑的现象、原因、解决复现源码和真正用起来是两个难度等级。下面这 5 个坑按出现频率排前三个我在本地环境反复踩后两个在联调阶段暴露。每条按现象、原因、解决三段式写方便你对日志定位。5.1 伪分布式下 Spark 任务频繁 OOM现象提交任务后日志出现 Container killed by YARN for exceeding memory limits任务反复重试然后失败。 原因一台机器上 YARN 声明了可用内存Spark 的 executor、driver 和 ApplicationMaster 全要从这个池子里分总和超过 NodeManager 声明的 4GB 时YARN 强制杀容器。 解决先盘点资源再调参。NodeManager 声明 8GB 时executor 给 2g、driver 给 1g、AM 默认 1g三个加起来小于 8g并且要留 20% 给操作系统。另外一个隐藏参数是 spark.executor.memoryOverhead默认是 executor 内存的 10%伪分布式下同样占实际内存。我当时在这上面折腾了两天最后整理了一张资源清单约束所有任务再没犯过。5.2 Spark 读取 JSON 时类型推断翻车现象printSchema 显示金额字段是 string日期字段全为 null模型训练报 IllegalArgumentException。 原因Spark 的 inferSchema 遇到字段里混着数字和空串时会把整个字段推断成 string日期格式不统一也会解析失败返回 null而且不报错。 解决显式声明 schema 再读from pyspark.sql.types import StructType, StructField, StringType, DoubleType, DateType schema StructType([ StructField(loan_id, StringType(), True), StructField(amount, DoubleType(), True), StructField(due_dt, DateType(), True), ]) df spark.read.schema(schema).json(hdfs://localhost:9000/risk/input/loan_*.json)注意 DateType 对格式要求严格不匹配时返回 null。我先用字符串读进来再统一 format 成标准格式可靠性最高。读的时候还可以加 option(mode, FAILFAST)遇到格式错误的记录直接抛错而不是静默把脏数据带进特征表。这条经验写在任何 spark 读取 json 的笔记里都不为过。5.3 NameNode 进入安全模式HDFS 只能读不能写现象向 HDFS 跑数据时报 org.apache.hadoop.ipc.RemoteException: Cannot create file但 ls 能正常列目录。 原因机器异常断电或直接 kill 进程后NameNode 检测到大量 block 未上报自动进入安全模式保证数据一致此时禁止写操作。 解决先执行 hdfs dfsadmin -report 查看 missing block 数量不是 0 就等待 DataNode 上报完成确认无误后执行 hdfs dfsadmin -safemode leave 手动退出。伪分布式下最常见的根因是 dfs.replication 配了 3 而只有 1 个 DataNodeblock 永远不满干脆把副本改成 1。这也是 Hadoop 面试题里安全模式最常考的考点退出条件是上报的 block 数量达到阈值默认 0.999。5.4 YARN 容器数超限任务秒失败现象spark-submit 提交后十秒内失败日志提示 requested resource 超过 maximum allocation。 原因yarn.scheduler.maximum-allocation-mb 配得太小Spark 申请的内存超过上限被直接拒绝跟业务代码没关系。 解决调 yarn-site.xml 里的两个参数然后重启 ResourceManagerproperty nameyarn.scheduler.maximum-allocation-mb/name value4096/value /property property nameyarn.scheduler.minimum-allocation-mb/name value512/value /property这个坑和 5.1 的区别是5.1 是容器跑着跑着被杀5.4 是根本排不上。排查时先看 ResourceManager 日志里的 allocation 数值再决定调哪边。很多人只改 spark.executor.memory 不生效因为 yarn-site 的上限才是硬约束。5.5 中文乱码与敏感列明文落库现象从业务方拿的 CSV 文件读进 Spark 后中文显示乱码结果表里手机号明文可查。 原因业务库导出的 CSV 默认 GBK 编码Spark 默认按 UTF-8 解码敏感列没做脱敏就写回业务表。 解决读 CSV 时显式指定编码df spark.read.option(encoding, GBK).csv(hdfs://localhost:9000/risk/input/user_info.csv)敏感列脱敏遵循“输出即脱敏”原则写入任何结果表前对手机号、身份证、联系地址统一打码不要依赖文件系统权限兜底。HDFS 上能读文件的人一多权限就是虚设。完整的权限设计要落在两层目录 ACL 控谁能读文件Spark SQL 列级脱敏控谁能看见敏感字段两层都做了才叫闭环。这个问题通常到联调阶段才暴露影响面大建议一开始就写进特征加工链路里。6. 从伪分布式到生产集群HA 配置、distcp 数据迁移与结果验证6.1 Hadoop HA 与 ZooKeeper 整合换集群时业务代码要改什么伪分布式验证完代码上生产集群前文件系统入口要从单个 NameNode 换成 HA 服务。常见做法是两个 NameNode 加三个 JournalNodeZooKeeper 负责自动 failover。core-site.xml 里的核心是 nameservices 配置property namedfs.nameservices/name valueriskcluster/value /property property namedfs.ha.namenodes.riskcluster/name valuenn1,nn2/value /property property namedfs.namenode.rpc-address.riskcluster.nn1/name valuehost1:8020/value /property property namedfs.namenode.rpc-address.riskcluster.nn2/name valuehost2:8020/value /property配置完成后客户端访问路径从 hdfs://localhost:9000 改成 hdfs://riskclusterSpark 业务代码不用动只改 spark-defaults.conf 里的 fs.defaultFS。如果还没走到 Spark 集群搭建这一步这套 HA 配置就是最值得提前看的预案。ZooKeeper 整合的坑主要集中在 fencing老 NameNode 没被真正杀掉会产生脑裂Hadoop 默认用 sshfence 确保老节点下线SSH 免密必须提前配好否则切换不成功。HA 的原理不复杂但没让一台机器直接断电验证过自动切换就不算真正做完。6.2 历史数据迁移distcp 参数与回读验证历史数据从旧集群迁到新集群用 distcp 而不是直接 cp。distcp 本质是 MapReduce 作业能并发、能断点续跑hadoop distcp -m 8 -bandwidth 50 -update -skipcrccheck \ hdfs://oldcluster:8020/risk/input \ hdfs://riskcluster/risk/input参数含义-m 指定并发 map 数生产上我很少超过 20太多会把源集群读爆-bandwidth 限制单 map 带宽单位是 MB/s大数据量迁移要错峰跑-update 表示源文件更新或目标缺失才复制适合增量-skipcrccheck 跳过 CRC 校验源数据量大时省时间但降低一点完整性保障。迁移完必须验证不能只看任务返回 success。两个检查方法hdfs dfs -du -s 对比源和目标的目录大小再用 Spark 读新路径做 count 和分区边界检查。我最早做迁移时图快把 -m 开到 64直接把源集群读满NameNode 响应延迟飙到秒级在线查询全被拖垮。后来养成的习惯是先小并行度跑十分钟看监控确认源集群负载平稳再加并发。这也是大数据集群部署策略里最容易被忽略的一条迁移不是把数据搬完而是保证搬的过程不影响正在跑的任务。希望帮到你。本文还有配套的精品资源点击获取