简介本资源是一份面向金融行业数据平台建设者、大数据工程师与风控系统架构师的「数据中台-数据决策平台」全景式技术方案PPT聚焦于如何构建以数据驱动业务创新为核心的实时化、可溯化决策支撑体系。内容涵盖系统定位与演进历程V1.0至V3.0、分层技术架构含规则引擎、机器学习模型、Kafka异步接入、流式计算引擎等20核心组件、四大功能模块指标管理、规则配置、流程调度、数据采集及典型业务成效如浙里贷、小微易贷等场景年均千万级模型调用。资源为单个1015KB的PPTX文件结构清晰、图示丰富包含系统架构图、决策流配置逻辑、指标来源矩阵及规则引擎组件分解便于快速掌握平台设计思想与落地要点。已有182人学习下载适合需复用架构设计、理解规则决策链路或开展同类平台建设的技术人员深度研读。1. 这不是又一个BI看板它能在3秒内把“客户刚在社保云查过公积金”变成“拒绝授信”的决策动作你有没有遇到过这种场景风控同事半夜发来截图说某笔贷款申请被拒但系统日志里只有一行decision_result: REJECT没人知道是哪条规则、哪个指标、哪次外部调用触发了这个结果或者市场部急着上线“公积金缴存满24个月送额度”活动技术要排期两周改代码、测环境、走发布流程——而业务方只想明天早上就能在后台点几下就生效。这个「数据中台-数据决策平台」就是专治这类顽疾的它把原本散落在数仓脚本、Java服务、Excel规则表里的决策逻辑全部收束到一套可视化配置体系里让业务人员能自己拖拽规则包、绑定指标、设置分支条件一次配置全链路实时生效。它不替代数仓建模也不取代机器学习模型训练而是做那个“翻译官”和“调度员”——把业务语言比如“近3个月征信查询超5次且无公积金缴存记录”精准翻译成可执行的计算指令并在毫秒级完成从Kafka拉取实时设备指纹、调用百融API、查ECIF配偶黑名单、跑ILOG规则引擎、写回信贷系统这一整套动作。项目正文里那句“总计在用的超过1000条指标”不是虚数而是真实压在生产环境上的决策原子而“浙里贷营销数字化授信模型计算9万余次/天”背后是这套平台扛住了每秒200并发的决策请求洪峰。如果你正被“业务提需求→开发改代码→测试卡流程→上线等排期”的死循环折磨或者手头有几十个信贷/营销/预警类模型却苦于无法统一纳管、版本混乱、溯源困难那它不是可选项而是你现在最该拆开跑起来的工程。2. 跑通第一单从源码解压到触发一笔真实决策请求含完整命令链提示本章所有操作均基于项目正文明确列出的技术组件——Kafka集群、流立方、ILOG引擎、ECIF/数仓对接能力均为平台原生支持无需额外魔改。我们跳过“下载JDK”“装Docker”这类通用前置直击平台启动核心路径。2.1 环境准备三件套必须齐缺一不可平台对运行环境有明确依赖不是“Java有就行”而是版本和组件功能强绑定。我按项目正文架构图里标出的“大数据平台计算引擎”“流式计算引擎”“Kafka集群”三个关键层梳理出最低可行组合JDK 11.0.15必须用LTS版本项目中pom.xml里maven-compiler-plugin明确指定source11用JDK17会触发UnsupportedClassVersionError血泪经验某次升级后所有规则包加载失败查了6小时才发现是JDK版本漂移Kafka 2.8.1集群3节点项目正文多次强调“Kafka集群异步接入”且application-prod.yml中kafka.bootstrap-servers配置格式为host1:9092,host2:9092,host3:9092。注意必须启用auto.create.topics.enabletrue否则首次启动时decision-result-topic和># 验证JDK版本必须11.0.15 java -version | grep 11\. | head -1 | awk {print $NF} | grep -q 11\.0\.1[5-9] echo SUCCESS || echo FAIL # 验证Kafka连通性替换your-kafka-host为实际地址 echo test | kafka-console-producer.sh --bootstrap-server your-kafka-host:9092 --topic test-topic 2/dev/null \ kafka-console-consumer.sh --bootstrap-server your-kafka-host:9092 --topic test-topic --from-beginning --max-messages 1 2/dev/null | grep -q test echo SUCCESS || echo FAIL # 验证MySQL连接替换user/pass/host mysql -h your-mysql-host -u user -ppass -e SELECT VERSION() LIKE 5.7.%; 2/dev/null | grep -q 1 echo SUCCESS || echo FAIL2.2 源码结构解析别急着mvn clean install先看清这五个核心模块项目解压后目录结构并非扁平化而是严格按“数据采集层→计算层→决策层→应用层”分层。我直接贴出生产环境部署时真正起作用的模块树删减了doc/、mock-data/等非运行目录data-decision-platform/ ├──>cd>cd ../metrics-service mvn clean package -Dmaven.test.skiptrue java -jar target/metrics-service-1.0.0.jar --spring.profiles.activeprod验证curl http://localhost:8082/api/v1/metrics/list?categoryinternal应返回JSON数组含CRM数仓、ECIF等指标源。启动decision-schedulercd ../decision-scheduler mvn clean package -Dmaven.test.skiptrue java -jar target/decision-scheduler-1.0.0.jar --spring.profiles.activeprod关键日志Started DecisionSchedulerApplication in X seconds后必见Initializing Kafka producer for topic: decision-result-topic。启动web-admincd ../web-admin npm install npm run build # 构建前端静态资源 mvn clean package -Dmaven.test.skiptrue java -jar target/web-admin-1.0.0.jar --spring.profiles.activeprod访问http://localhost:8080输入默认账号admin/admin123登录。若首页显示“指标总数1024”说明全链路打通。2.4 发起第一笔决策请求用curl模拟信贷系统调用现在平台已活我们绕过前端用最原始的HTTP请求触发一次完整决策流。项目正文描述的“业务系统发起授信申请到调度平台”对应的就是这个接口。curl -X POST http://localhost:8081/api/v1/schedule/loan \ -H Content-Type: application/json \ -d { applyId: TEST20231001001, customerId: CUST123456, productCode: ZHELI_DAI, applyAmount: 50000, applyTerm: 12, externalDataSources: [ecif, credit_report, housing_fund] }预期响应200 OK{ code: 200, message: 决策调度成功, data: { scheduleId: SCH2023100100001, status: PROCESSING, startTime: 2023-10-01T10:30:45.123 } }验证是否真跑通查看decision-scheduler日志应出现类似[DecisionFlowExecutor] - Starting flow for applyId: TEST20231001001, product: ZHELI_DAI [DataCollectClient] - Calling ECIF API for customer: CUST123456 [KafkaProducer] - Sent result to topic: decision-result-topic, key: TEST20231001001检查Kafka中decision-result-topic是否有消息kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic decision-result-topic \ --from-beginning \ --max-messages 1 \ --value-deserializer org.apache.kafka.common.serialization.StringDeserializer输出应为JSON含applyId:TEST20231001001及decisionResult:APPROVE或REJECT。至此你已亲手驱动了从信贷系统请求→调度引擎分发→数据采集→规则计算→结果回传的全链路。这不是Demo而是生产级决策流的最小可行闭环。3. 核心模块代码深挖看懂DecisionFlowExecutor如何把规则包变成可执行字节码提示本章代码均来自decision-scheduler模块的DecisionFlowExecutor.java这是项目正文里“调度平台通知数据采集清洗引擎进行数据采集和清洗操作”这一动作的实际执行者。我们不讲设计模式只看它怎么把XML规则变成CPU指令。3.1 决策流执行入口executeFlow()方法的三段式结构打开DecisionFlowExecutor.javaexecuteFlow()是整个调度的起点。它不是简单地for循环执行规则而是分三阶段处理对应项目正文“调度平台通知→采集引擎获取→返回指标数据”这一链条public DecisionResult executeFlow(LoanApplyRequest request) { // 阶段1准备上下文 —— 构建决策所需的初始数据容器 DecisionContext context buildInitialContext(request); // ← 关键注入applyId、customerId等基础字段 // 阶段2执行规则流 —— 按配置顺序调用各节点服务 try { for (FlowNode node : flowConfig.getNodes()) { // flowConfig来自数据库的流程配置 context nodeService.execute(node, context); // ← 核心每个node对应一个微服务调用 } } catch (Exception e) { log.error(Flow execution failed at node: {}, node.getName(), e); throw new DecisionException(FLOW_EXECUTION_ERROR, e); } // 阶段3生成最终结果 —— 将context中所有指标汇总为决策结论 return generateFinalResult(context); // ← 返回APPROVE/REJECT及明细指标 }参数说明LoanApplyRequest就是上一章curl里传的JSON反序列化对象productCode字段决定走哪套规则流ZHELI_DAI对应浙里贷流程DecisionContext贯穿全程的“数据背包”内部是MapString, Object键为指标名如ecif.creditLimit值为具体数值或对象flowConfig.getNodes()从flow_config表查出的节点列表例如[{type:DATA_COLLECT,service:data-collect-service}, {type:RULE_EVAL,service:rule-engine-core}]为什么这样设计项目正文强调“结合场景特征的灵活可视化配置”如果把所有逻辑硬编码在executeFlow()里每次新增产品如“分期贷”都要改Java代码。而三段式结构让flowConfig成为真正的配置中心——运维只需在后台修改flow_config表新增一个{type:RISK_SCORE,service:risk-model-service}节点调度引擎下次执行时自动调用新服务零代码变更。3.2 数据采集节点DataCollectNodeService如何对接12类数据源nodeService.execute()中当node.type DATA_COLLECT时实际调用的是DataCollectNodeService。它不是写死调用ECIF或百融而是通过策略模式动态选择实现类Component public class DataCollectNodeService implements NodeService { Autowired private MapString, DataCollector collectorMap; // ← Spring自动注入所有DataCollector实现 Override public DecisionContext execute(FlowNode node, DecisionContext context) { String dataSource node.getParams().get(dataSource); // 从流程配置中读取如credit_report DataCollector collector collectorMap.get(dataSource); // ← 关键根据dataSource字符串找Bean if (collector null) { throw new IllegalArgumentException(No collector found for dataSource: dataSource); } // 执行采集传入context含customerId返回新指标 MapString, Object newMetrics collector.collect(context); context.putAll(newMetrics); // 合并到决策上下文 return context; } }collectorMap里有哪些实现项目正文提到“行外数据百融、地税、个人征信、公积金...”对应>Component public class RuleEngineClient { Value(classpath:rule/) // ← 对应2.2节说的rule/目录 private Resource ruleDir; private final RulePackageLoader loader new RulePackageLoader(); public DecisionResult evaluate(DecisionContext context) { // 1. 根据productCode加载对应规则包如ZHELI_DAI.xml RulePackage rulePackage loader.loadByProduct(context.getProductCode()); // 2. 创建执行环境传入context作为变量池 RuleExecutionContext env new RuleExecutionContext(context); // 3. 执行底层用Janino编译器将XML中的condition表达式转为Java字节码 Object result rulePackage.execute(env); return DecisionResult.builder() .applyId(context.getApplyId()) .decisionResult((String) result) // ← XML中action标签的返回值 .build(); } }看一个真实的ZHELI_DAI.xml片段rulePackage nameZHELI_DAI version2.1 rule name公积金缴存校验 priority10 condition ![CDATA[ context.get(housing_fund.monthsSinceLastContribution) ! null ((Integer)context.get(housing_fund.monthsSinceLastContribution)) 24 ]] /condition actionAPPROVE/action /rule rule name征信查询超频 priority20 condition ![CDATA[ context.get(credit_report.inquiryCountLast3Months) ! null ((Integer)context.get(credit_report.inquiryCountLast3Months)) 5 ]] /condition actionREJECT/action /rule /rulePackage关键点condition里的context.get()调用就是DecisionContext的Map访问所以数据采集节点必须把指标put进去规则才能取到priority决定执行顺序项目正文“多个规则包或多个规则组合成规则组”就是靠这个数字排序Janino编译器把CDATA里的字符串当成Java代码编译所以类型强转((Integer)...必不可少否则运行时报ClassCastException3.4 结果持久化KafkaResultPublisher如何保证“至少一次”投递决策完成后generateFinalResult()会调用KafkaResultPublisher将结果发往Kafka。项目正文说“调度平台将决策结果返回到业务系统端并将加工好的指标数据、决策结果推送至kafka队列”这个类就是执行后者Component public class KafkaResultPublisher { Autowired private KafkaTemplateString, String kafkaTemplate; Value(${kafka.topic.result}) private String resultTopic; public void publish(DecisionResult result) { // 1. 构建消息key为applyId保证同一申请的所有结果路由到同一partition ProducerRecordString, String record new ProducerRecord( resultTopic, result.getApplyId(), // ← 关键分区键 JSON.toJSONString(result) ); // 2. 异步发送 回调处理失败 ListenableFutureSendResultString, String future kafkaTemplate.send(record); future.addCallback( success - log.info(Sent to Kafka: {}, result.getApplyId()), failure - { log.error(Failed to send to Kafka, failure); // ← 项目正文“数据等待机制避免数据漏过风险”的实现 // 失败时写入本地failed_result表由定时任务重试 failedResultRepository.save(new FailedResult(result)); } ); } }为什么用applyId作key项目正文提到“贷中预警计算9万余次”意味着同一客户可能有多笔申请并发决策。用applyId作key确保TEST20231001001和TEST20231001002一定进入不同partition避免因单partition积压导致所有申请延迟。而failed_result表的存在正是应对网络抖动的“后悔药”——我见过生产环境因Kafka集群滚动重启导致17分钟内327条结果发送失败全靠这个表兜底重试。4. 避坑指南那些让老司机也翻车的5个致命细节提示以下问题全部来自真实部署案例非理论推演。每一条都附带grep命令快速定位日志省去你查半天的冤枉路。4.1 现象调度引擎日志疯狂刷Failed to load rule package for product: ZHELI_DAI但rule/ZHELI_DAI.xml明明存在原因RulePackageLoader默认只扫描classpath:rule/下的XML而>resource directorysrc/main/resources/directory includes include**/*.xml/include include**/*.yml/include /includes /resource然后mvn clean package重打jar。验证命令jar -tf target/data-collect-service-1.0.0.jar | grep ZHELI_DAI.xml。4.2 现象web-admin登录后指标管理页空白浏览器F12看到GET /api/v1/metrics/list 500原因metrics-service的application.yml中spring.datasource.url配置了useSSLfalseserverTimezoneUTC但MySQL 5.7.32要求serverTimezoneGMT%2B8URL编码后。解决修改metrics-service/src/main/resources/application.ymlspring: datasource: url: jdbc:mysql://mysql-host:3306/metrics_db?useSSLfalseserverTimezoneGMT%2B8重启服务后执行curl -v http://localhost:8082/api/v1/metrics/list应返回200。4.3 现象Kafka中decision-result-topic有消息但信贷系统没收到回调decision-scheduler日志无异常原因项目正文“HTTP/Web Service/Socket消息处理”模块中HttpCallbackService默认只监听http://callback-service:8080/notify但信贷系统实际回调地址是https://loan-api.example.com/v1/decision/callback且需要Bearer Token认证。解决在decision-scheduler/src/main/resources/application-prod.yml中增加callback: url: https://loan-api.example.com/v1/decision/callback headers: Authorization: Bearer ${CALLBACK_TOKEN}并在启动时传入-DCALLBACK_TOKENabc123。验证curl -H Authorization: Bearer abc123 https://loan-api.example.com/v1/decision/callback应返回OK。4.4 现象规则配置界面保存“操作规则”后rule/目录下没生成XML且rule-engine-core日志无任何输出原因web-admin前端调用的是/api/v1/rule/save接口该接口在RuleController.java中但其PostMapping缺少consumes MediaType.APPLICATION_JSON_VALUE导致Spring MVC无法解析JSON body静默失败。解决修改web-admin/src/main/java/com/xxx/controller/RuleController.javaPostMapping(value /save, consumes MediaType.APPLICATION_JSON_VALUE) public Result save(RequestBody RuleSaveRequest request) { ... }重新构建前端并重启web-admin。4.5 现象流立方StreamCube实时指标无法被采集>streamcube: host: streamcube.default.svc.cluster.local # ← K8s环境必须用FQDN port: 8080若在物理机部署则改为实际IPstreamcube.host: 192.168.1.100。5. 二次开发实战给“小微易贷”产品加一个“企业纳税额连续3年增长”规则提示本章不讲“如何写Java”而是聚焦“如何在不改核心代码的前提下让新规则上线”。项目正文强调“支持行社个性化产品模型的配置”这就是它的落地姿势。5.1 新增指标从“企业纳税额”原始数据到可配置指标项目正文指标体系里有“行内数据CRM数仓、ECIF”但没提“企业纳税额”。我们要先把它变成平台可识别的指标步骤如下在数仓中确认数据源找DBA确认企业纳税数据在dw_corp_tax表中关键字段corp_id企业ID、tax_year年份、tax_amount纳税额。项目正文“数仓每日进行指标批处理加工任务”意味着此表每天凌晨ETL更新。在metrics-service中注册新指标编辑metrics-service/src/main/resources/metrics-definition.json添加{ metricCode: corp.tax.amount.last3years, metricName: 企业纳税额最近3年, category: corporate, sourceType: DW, sourceTable: dw_corp_tax, sqlTemplate: SELECT tax_amount FROM dw_corp_tax WHERE corp_id ? AND tax_year IN (?, ?, ?), params: [corpId, year1, year2, year3] }注意sqlTemplate必须用?占位符params顺序必须与SQL中?位置一致这是MetricsService执行查询的契约。重启metrics-service并验证curl http://localhost:8082/api/v1/metrics/get?metricCodecorp.tax.amount.last3yearscorpIdCORP789year12021year22022year32023应返回JSON数组[125000, 138000, 152000]。5.2 配置衍生指标把“连续3年增长”抽象成布尔型指标项目正文“衍生指标主要在基础指标的基础上进行简单的四则运算”但“连续增长”是逻辑判断需用Groovy脚本在metrics-service中添加Groovy脚本创建metrics-service/src/main/resources/groovy/corp_tax_growth.groovy// 输入taxAmounts [125000, 138000, 152000] def isGrowing true for (int i 1; i taxAmounts.size(); i) { if (taxAmounts[i] taxAmounts[i-1]) { isGrowing false break } } return isGrowing在metrics-definition.json中注册衍生指标{ metricCode: corp.tax.growth.3years, metricName: 企业纳税额连续3年增长, category: corporate, sourceType: GROOVY, scriptPath: groovy/corp_tax_growth.groovy, dependsOn: [corp.tax.amount.last3years] }dependsOn声明了依赖关系MetricsService会先查corp.tax.amount.last3years再把结果传给Groovy脚本。验证衍生指标curl http://localhost:8082/api/v1/metrics/get?metricCodecorp.tax.growth.3yearscorpIdCORP789year12021year22022year32023返回true或false。5.3 在规则包中引用新指标修改XIAO_WEI_YI_DAI.xml找到rule/XIAO_WEI_YI_DAI.xml在rulePackage内添加新规则rule name企业纳税额连续3年增长校验 priority5 condition ![CDATA[ context.get(corp.tax.growth.3years) ! null ((Boolean)context.get(corp.tax.growth.3years)) ]] /condition actionAPPROVE/action /rule注意priority5设为最高优先级确保此规则最先执行。项目正文“快速发布实现全流程各节点项实时监控”就是靠这种低侵入式修改。5.4 给“小微易贷”产品授权该指标项目正文“新增指标授权到产品级”需在数据库中插入授权记录INSERT INTO metric_authorization (product_code, metric_code, status) VALUES (XIAO_WEI_YI_DAI, corp.tax.growth.3years, ENABLED);5.5 验证全流程从申请到决策结果发起一笔针对企业的申请请求curl -X POST http://localhost:8081/api/v1/schedule/loan \ -H Content-Type: application/json \ -d { applyId: XWYD20231001001, customerId: CORP789, productCode: XIAO_WEI_YI_DAI, applyAmount: 1000000, applyTerm: 24, externalDataSources: [ecif] }检查Kafka结果kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic decision-result-topic \ --from-beginning \ --max-messages 1 \ --value-deserializer org.apache.kafka.common.serialization.StringDeserializer | \ jq .decisionResult若返回APPROVE且日志中出现[RuleEngineClient] - Executed rule: 企业纳税额连续3年增长校验则大功告成。从那以后我每次给新业务线加规则都强制走一遍“指标注册→衍生定义→规则编写→产品授权→curl验证”五步哪怕只是改个数字阈值。因为项目正文里那句“支持行社个性化产品模型的配置”不是口号而是刻在metrics-definition.json和rule/目录结构里的契约——只要守住这个契约你就能在业务需求爆发时把上线周期从两周压缩到两小时。希望帮到你。本文还有配套的精品资源点击获取