Hadoop+Spark信贷风控系统实战：从环境搭建到窗口函数特征工程-北京尧图网络

简介这是一套面向计算机相关专业学生与项目实战学习者的毕业设计级资源主题为基于Hadoop与Spark的大数据金融信贷风险控制系统适合用作毕设、课程设计或期末大作业也可作为大数据方向求职者的练手项目。压缩包共69个文件约59KB以36个Java源文件与8个Scala脚本构成核心业务与Spark计算逻辑辅以12个XML配置、5个properties参数文件及SQL建表脚本、JSON数据样例、Markdown说明文档等覆盖数据源接入、流式处理与信贷风控模块。项目经导师指导并认可已严格调试可运行目录中可见h5-credit-risk-control前端、data-source-spark-streaming流处理与credit-risk-control后端等模块便于理解HadoopSpark在金融商贸场景下的工程组织方式。目前已有293人学习适合需要完整赛题方案与排错思路的读者参考。1. 从一份信贷风控源码说起HadoopSpark 这套组合到底能跑出什么信贷风控这个场景数据量从来不是小数目。一个中等规模的消费金融平台每天产生的借款申请、还款记录、逾期行为、征信查询日志加起来轻松上千万条用单机 MySQL 做特征计算和模型训练光一个用户行为画像的聚合就能把磁盘 IO 打满。这份基于 HadoopSpark 的大数据金融信贷风险控系统源码解决的就是这个量级下的数据处理和风险指标计算问题。它把离线批处理和实时流处理两条链路都搭好了Hadoop 负责底层分布式存储和海量历史数据的批处理Spark 负责内存计算加速特征工程和风险评分前端用 H5 做展示层后端 Java 服务做接口聚合。适合正在做大数据方向毕业设计的学生也适合想找一个完整项目练手 Spark 实战的开发者。源码包里带了项目说明和数据库脚本拿到之后按步骤配环境就能跑起来。2. 环境搭建与数据链路打通从 Hadoop 伪分布式到 Spark 跑通第一条风控指标2.1 为什么选 Hadoop 伪分布式而不是单机模式这套源码的存储层依赖 HDFS计算层依赖 Spark on YARN 或者 Spark Standalone。如果你只是本地跑个 demo用 Hadoop 单机模式也能读写本地文件系统但源码里很多路径写的是hdfs://开头的 URI单机模式下会直接报Wrong FS错误。常见做法是搭一个伪分布式环境NameNode、DataNode、ResourceManager、NodeManager 都跑在同一台机器上配置改起来不复杂但能完整模拟分布式文件系统的行为。伪分布式搭建的核心改动集中在三个配置文件core-site.xml指定 HDFS 的默认 FS 地址hdfs-site.xml设置副本数为 1伪分布式只有一个 DataNode设 3 会一直报副本不足mapred-site.xml和yarn-site.xml把计算框架指向 YARN。下面是我一般会用的最小配置片段!-- core-site.xml -- configuration property namefs.defaultFS/name valuehdfs://localhost:9000/value /property property namehadoop.tmp.dir/name value/opt/module/hadoop-3.x/data/tmp/value /property /configuration !-- hdfs-site.xml -- configuration property namedfs.replication/name value1/value /property property namedfs.namenode.name.dir/name value/opt/module/hadoop-3.x/data/namenode/value /property property namedfs.datanode.data.dir/name value/opt/module/hadoop-3.x/data/datanode/value /property /configurationfs.defaultFS里的端口 9000 是 NameNode 的 RPC 端口如果你机器上 9000 被占用了比如某些监控 agent改成 8020 也行但后面 Spark 提交任务时--master yarn读的就是这个配置改了一处就要全局一致。hadoop.tmp.dir建议单独挂一块数据盘不要放在/tmp下系统重启后/tmp被清空会导致 NameNode 元数据丢失集群直接起不来这个坑我踩过不止一次。格式化 NameNode 的命令是hdfs namenode -format注意这个操作只能执行一次。如果你后续改了dfs.namenode.name.dir的路径需要先把旧数据目录清掉再重新格式化否则会出现 clusterID 不一致导致 DataNode 拒绝注册。启动顺序也有讲究先start-dfs.sh再start-yarn.sh最后用jps确认 NameNode、DataNode、ResourceManager、NodeManager 四个进程都在。2.2 Spark 读取 HDFS 数据做信贷指标聚合环境起来之后下一步是把源码里的数据源接入 Spark。这份项目的数据源分两块一块是>from pyspark.sql import SparkSession from pyspark.sql.functions import col, count, sum, when, datediff, current_date spark SparkSession.builder \ .appName(CreditRiskBatch) \ .config(spark.sql.warehouse.dir, /user/hive/warehouse) \ .enableHiveSupport() \ .getOrCreate() # 读取借款申请明细假设已同步到 HDFS 的 /data/credit/apply 路径 apply_df spark.read.parquet(hdfs://localhost:9000/data/credit/apply) # 计算每个用户近 90 天的多头借贷次数和逾期天数 risk_df apply_df.filter(datediff(current_date(), col(apply_date)) 90) \ .groupBy(user_id) \ .agg( count(loan_id).alias(loan_cnt_90d), sum(when(col(overdue_days) 0, 1).otherwise(0)).alias(overdue_cnt_90d), sum(overdue_days).alias(total_overdue_days) ) \ .withColumn(risk_level, when((col(loan_cnt_90d) 5) | (col(total_overdue_days) 30), HIGH) .when((col(loan_cnt_90d) 2) | (col(total_overdue_days) 7), MEDIUM) .otherwise(LOW)) risk_df.write.mode(overwrite).parquet(hdfs://localhost:9000/data/credit/risk_result)这段代码的逻辑不复杂但有几个参数需要根据实际数据调整。datediff(current_date(), col(apply_date)) 90这个窗口期决定了你观察多头借贷的时间范围信贷场景里通常用 90 天但如果你做的是现金贷短期产品改成 30 天更合理。loan_cnt_90d 5这个阈值也不是固定的源码里给的是一个经验值实际业务中需要根据通过率和坏账率的平衡来调。risk_level的三档分类直接对应前端 H5 页面上的风险标签展示如果你改了档位名称记得同步改credit-risk-control模块里的枚举类。写 Parquet 的时候用mode(overwrite)会全量覆盖目标路径如果你要做增量更新改成mode(append)并加上分区字段。但注意 Parquet 的 append 模式不会去重同一批数据跑两次会出现重复记录生产环境一般会按日期分区后覆盖当天分区。2.3 流式数据源接入与 Spark Streaming 消费># 启动 Socket 模拟数据源测试用 nc -lk 9999 # 另开终端提交 Spark Streaming 任务 spark-submit \ --class com.credit.risk.streaming.RiskStreamingApp \ --master local[2] \ --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 \ /path/to/credit-risk-control.jar--master local[2]表示本地两个线程跑测试够用了。--packages那行是引入 Kafka 连接器如果你用的是 Socket 方式就不需要。提交之前确认credit-risk-control.jar已经用 Maven 打过包pom.xml里 Spark 的 scope 是provided打包时不会把 Spark 依赖塞进去这是对的不然 jar 包会大到离谱。流处理的核心逻辑是每 5 秒一个微批从 Socket 或 Kafka 读取 JSON解析出user_id、apply_amount、apply_time字段然后和离线算好的风险画像做 join实时输出风险评分。这里有个坑流数据和离线画像的 join 如果画像表太大每个微批都去查一次 HDFS 会非常慢。常见做法是把画像表缓存到内存或者 Redis 里Spark Streaming 里用foreachRDD配合广播变量来减少重复读取。3. 风控指标计算与 H5 展示层对接从 Spark 结果表到前端可读的 JSON3.1 风险指标的计算口径与 Spark SQL 实现信贷风控的核心指标就那么几个逾期率DPD、授信使用率、多头借贷指数、收入负债比。源码里credit-risk-control模块的GeneratorMapper.xml定义了这些指标的 SQL 查询但它是基于 MySQL 的数据量大了跑不动。我一般会把同样的逻辑用 Spark SQL 重写一遍跑在 HDFS 的 Parquet 数据上。-- 逾期率按账龄分桶计算 SELECT CASE WHEN overdue_days 0 THEN M0 WHEN overdue_days BETWEEN 1 AND 30 THEN M1 WHEN overdue_days BETWEEN 31 AND 60 THEN M2 WHEN overdue_days BETWEEN 61 AND 90 THEN M3 ELSE M3 END AS bucket, COUNT(DISTINCT user_id) AS user_cnt, SUM(loan_amount) AS total_amount, AVG(overdue_days) AS avg_overdue_days FROM credit_loan_detail WHERE stat_date ${biz_date} GROUP BY CASE WHEN overdue_days 0 THEN M0 WHEN overdue_days BETWEEN 1 AND 30 THEN M1 WHEN overdue_days BETWEEN 31 AND 60 THEN M2 WHEN overdue_days BETWEEN 61 AND 90 THEN M3 ELSE M3 END这个分桶逻辑和银行的标准口径基本一致M0 是正常M1 是逾期 1-30 天M2 是 31-60 天M3 是 61-90 天M3 是超过 90 天。stat_date是跑批日期一般用 T-1 的数据。COUNT(DISTINCT user_id)在 Spark SQL 里会触发 shuffle数据量大的时候比较慢如果不需要精确去重可以用approx_count_distinct替代误差在 2% 以内速度能快好几倍。3.2 后端接口如何把 Spark 结果吐给 H5 前端H5 前端在h5-credit-risk-control目录下用的是 Vue 或者 React看package-lock.json里的依赖通过 HTTP 请求后端接口拿数据。后端credit-risk-control模块是一个 Spring Boot 应用pom.xml里引了 MyBatis 和 MySQL 驱动。但 Spark 算出来的结果在 HDFS 上Spring Boot 直接读 HDFS 不太方便常见做法是 Spark 跑完之后把结果表同步到 MySQL后端只查 MySQL。# Spark 结果导出到 MySQL spark-submit \ --class com.credit.risk.export.MysqlExporter \ --master yarn \ --deploy-mode client \ --jars /opt/module/mysql-connector-java-8.0.30.jar \ /path/to/credit-risk-control.jar \ --input hdfs://localhost:9000/data/credit/risk_result \ --output jdbc:mysql://localhost:3306/credit_risk \ --table risk_summary--jars指定 MySQL 驱动包--input是 HDFS 上的 Parquet 路径--output是 JDBC 连接串。导出的时候注意 MySQL 表的字符集要设成utf8mb4不然中文标签会乱码。另外 Spark 写 MySQL 默认是追加模式如果你要覆盖需要在代码里先TRUNCATE TABLE再写或者用mode(overwrite)配合truncate选项。后端接口的 Controller 层拿到 MySQL 里的数据后组装成 JSON 返回给前端。这里有个细节H5 页面上的风险等级标签高/中/低和 Spark 算出来的risk_level字段要一一对应如果前端写死了颜色映射后端改字段名的时候前端会直接白屏。我一般会在接口层加一层 DTO 转换把数据库字段映射成前端约定的枚举值这样两边解耦。3.3 前端展示与数据刷新频率的取舍H5 页面上的风险看板一般展示几个核心数字今日申请量、通过率、逾期率、高风险用户占比。这些数字的刷新频率取决于你的流处理延迟。如果 Spark Streaming 的微批间隔是 5 秒前端轮询间隔设 10 秒就够了设太快没意义还浪费请求。源码里前端默认是 30 秒轮询一次对于毕业设计演示来说够用了但如果你要模拟实时风控大屏改成 WebSocket 推送体验会好很多。注意前端轮询接口的时候记得加防抖不然用户快速切换页面会瞬间发出几十个请求后端连接池直接被打满。4. 避坑与排查这套源码跑不起来时先看这几个地方4.1 启动报 ClassNotFoundException: com.mysql.cj.jdbc.Driver现象是 Spring Boot 启动直接挂掉日志里一堆ClassNotFoundException。原因是pom.xml里 MySQL 驱动的 scope 被设成了provided打包时没打进去。解决方法是把 scope 改成runtime或者直接删掉 scope 标签重新mvn clean package。如果你用的是 IDEA 直接跑检查一下Run Configurations里有没有勾选Include dependencies with Provided scope。4.2 Spark 任务提交后一直卡在 ACCEPTED 状态现象是spark-submit之后 YARN 的 ResourceManager 页面显示应用状态一直是 ACCEPTED不变成 RUNNING。原因通常是 YARN 的队列资源不够或者yarn.nodemanager.resource.memory-mb设得太小容器申请不到内存。解决方法是去yarn-site.xml里把 NodeManager 的可用内存调大至少给 4G然后重启 YARN。另外检查一下spark-submit里的--executor-memory是不是超过了单节点可用内存伪分布式环境下设 1G 就够了。4.3 HDFS 写入报错 Could only be replicated to 0 nodes现象是 Spark 写 Parquet 到 HDFS 时抛异常说副本只能写到 0 个节点。原因是 DataNode 没起来或者dfs.replication设成了 3 但只有一个 DataNode。先jps确认 DataNode 进程在不在如果不在去看 DataNode 日志常见的是dfs.datanode.data.dir指向的目录权限不对或者磁盘满了。伪分布式环境下把dfs.replication改成 1 就能解决大部分问题。4.4 前端页面空白控制台报 404 或跨域错误现象是 H5 页面打开后一片空白F12 看 Network 里接口请求全是 404 或者 CORS 错误。原因是后端接口的 context-path 和前端配置的 baseURL 不一致或者后端没加跨域配置。检查application.yml里的server.servlet.context-path然后在前端.env文件里把VUE_APP_BASE_API改成一样的路径。跨域问题在 Spring Boot 里加一个CrossOrigin注解或者全局 CORS 配置就行注意allowedOrigins不要写*配合allowCredentials(true)浏览器会直接拒绝。4.5 流处理任务跑一段时间后 OOM现象是 Spark Streaming 任务跑几十分钟后 executor 挂掉日志里是java.lang.OutOfMemoryError。原因是流数据不断累积窗口操作没有设置水位线或者状态数据无限增长。检查代码里有没有用updateStateByKey或者mapWithState如果有必须设置spark.streaming.unpersist和合理的 batch interval。另外spark.streaming.kafka.maxRatePerPartition要设一个上限不然 Kafka 积压的时候一次性拉太多数据直接把内存撑爆。5. 进阶技巧用 Spark SQL 的窗口函数做用户行为序列特征前面几章把环境、指标计算和前后端对接都串起来了这一章说一个在实际风控建模里很实用但源码里没展开的技巧用 Spark SQL 的窗口函数从用户行为序列里提取时间间隔特征。信贷风控里有一个强特征叫「申请间隔」就是同一个用户两次借款申请之间隔了多少天。间隔特别短比如一天内申请五次大概率是急用钱或者多头借贷风险等级直接拉高。from pyspark.sql import Window from pyspark.sql.functions import lag, col, datediff, row_number # 按用户分组按申请时间排序取上一次申请时间 window_spec Window.partitionBy(user_id).orderBy(apply_time) apply_seq spark.read.parquet(hdfs://localhost:9000/data/credit/apply) \ .withColumn(prev_apply_time, lag(apply_time, 1).over(window_spec)) \ .withColumn(days_since_last_apply, datediff(apply_time, prev_apply_time)) \ .withColumn(apply_rank, row_number().over(window_spec)) # 筛选出 7 天内多次申请的用户 high_freq apply_seq.filter( (col(days_since_last_apply) 7) (col(apply_rank) 1) ).groupBy(user_id).count().filter(col(count) 3) high_freq.show(10)lag(apply_time, 1)取的是同一用户上一条申请记录的时间datediff算出间隔天数。row_number()给每个用户的申请记录编个号第一次申请 rank 是 1第二次是 2。最后筛出 7 天内申请超过 3 次的用户这批人就是高风险名单。这个逻辑在源码里没有现成的但你可以把它加到RiskBatch任务里算完之后把high_freq的结果写到 HDFS 的一个独立路径前端加一个「高频申请预警」的卡片来展示。窗口函数在 Spark SQL 里会触发 shuffle数据量大的时候建议先按user_id做一次repartition把同一个用户的数据分到同一个分区减少跨节点传输。另外lag函数在窗口边界会返回 nulldatediff遇到 null 也返回 null所以第一次申请的记录days_since_last_apply是空的过滤的时候要加isNotNull判断不然会报类型转换错误。还有一个细节apply_time字段如果是字符串格式比如2024-01-15 10:30:00datediff会隐式转换成 date 类型但只取日期部分同一天内的多次申请间隔会算成 0 天。如果你需要精确到小时用unix_timestamp转成秒再除以 3600。这个精度问题在源码的 SQL 脚本里没有体现但实际跑数据的时候会直接影响高频用户的判定结果。从那以后我每次拿到一份新的风控数据集都会先跑一遍lag窗口函数看看申请间隔的分布确认没有因为时间格式问题导致特征失真再往下做模型训练。希望帮到你。本文还有配套的精品资源点击获取

Hadoop+Spark信贷风控系统实战：从环境搭建到窗口函数特征工程

相关推荐

Python系统工具模块详解：从os、pathlib到subprocess的工程实践

基于Python的农作物病虫害识别分类项目：源码、数据集与迁移学习实战

TR 38.901-e30 信道模型实战：从参数解读到5G波束管理仿真落地

更多资讯

SAP云平台Cockpit使用指南：从子账户到服务实例的完整实操路径

Python工程化入门：面向对象、文件读写与异常处理实战笔记

R语言基础入门：向量、循环法则与NA/NaN处理

南开大学编译原理作业实战：简单C语言编译器从词法到代码生成

基于Docker的大学生兼职平台部署：容器化设计与避坑实践

AI数据合规全链路指南：从训练语料到模型输出的落地实践

今日更新

Susi AI API实战指南：susi_alexa_skill如何用Node.js调用chat.json获取智能回答

换新电脑了？KeyStats 恢复码数据找回完全指南，端到端加密统计一键重建

重建二叉树：由前序与中序遍历序列还原二叉树的原理与 Java 实现（YCBlogs 剑指 Offer 精讲）

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？