数据决策平台：业务规则可视化配置与实时决策执行

简介：本资源是一份面向金融行业数据平台建设者、大数据工程师与风控系统架构师的「数据中台-数据决策平台」全景式技术方案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明确指定source=11，用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.enable=true，否则首次启动时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 "FAIL"

    2.2 源码结构解析：别急着mvn clean install，先看清这五个核心模块

    项目解压后目录结构并非扁平化，而是严格按“数据采集层→计算层→决策层→应用层”分层。我直接贴出生产环境部署时真正起作用的模块树（删减了doc/、mock-data/等非运行目录）：

    data-decision-platform/ ├──>cd>cd ../metrics-service mvn clean package -Dmaven.test.skip=true java -jar target/metrics-service-1.0.0.jar --spring.profiles.active=prod

    验证：curl http://localhost:8082/api/v1/metrics/list?category=internal应返回JSON数组，含CRM数仓、ECIF等指标源。

  • 启动decision-scheduler

    cd ../decision-scheduler mvn clean package -Dmaven.test.skip=true java -jar target/decision-scheduler-1.0.0.jar --spring.profiles.active=prod

    关键日志：Started DecisionSchedulerApplication in X seconds后，必见Initializing Kafka producer for topic: decision-result-topic。

  • 启动web-admin

    cd ../web-admin npm install && npm run build # 构建前端静态资源 mvn clean package -Dmaven.test.skip=true java -jar target/web-admin-1.0.0.jar --spring.profiles.active=prod

    访问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.java，executeFlow()是整个调度的起点。它不是简单地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：贯穿全程的“数据背包”，内部是Map<String, 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 Map<String, 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），返回新指标 Map<String, 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 name="ZHELI_DAI" version="2.1"> <rule name="公积金缴存校验" priority="10"> <condition> <![CDATA[ context.get("housing_fund.monthsSinceLastContribution") != null && ((Integer)context.get("housing_fund.monthsSinceLastContribution")) >= 24 ]]> </condition> <action>APPROVE</action> </rule> <rule name="征信查询超频" priority="20"> <condition> <![CDATA[ context.get("credit_report.inquiryCountLast3Months") != null && ((Integer)context.get("credit_report.inquiryCountLast3Months")) > 5 ]]> </condition> <action>REJECT</action> </rule> </rulePackage>

    关键点：

    • <condition>里的context.get()调用，就是DecisionContext的Map访问，所以数据采集节点必须把指标put进去，规则才能取到
    • priority决定执行顺序，项目正文“多个规则包或多个规则组合成规则组”就是靠这个数字排序
    • Janino编译器把CDATA里的字符串当成Java代码编译，所以类型强转((Integer)...必不可少，否则运行时报ClassCastException

    3.4 结果持久化：KafkaResultPublisher如何保证“至少一次”投递

    决策完成后，generateFinalResult()会调用KafkaResultPublisher将结果发往Kafka。项目正文说“调度平台将决策结果返回到业务系统端；并将加工好的指标数据、决策结果推送至kafka队列”，这个类就是执行后者：

    @Component public class KafkaResultPublisher { @Autowired private KafkaTemplate<String, String> kafkaTemplate; @Value("${kafka.topic.result}") private String resultTopic; public void publish(DecisionResult result) { // 1. 构建消息：key为applyId，保证同一申请的所有结果路由到同一partition ProducerRecord<String, String> record = new ProducerRecord<>( resultTopic, result.getApplyId(), // ← 关键：分区键 JSON.toJSONString(result) ); // 2. 异步发送 + 回调处理失败 ListenableFuture<SendResult<String, 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> <directory>src/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配置了useSSL=false&serverTimezone=UTC，但MySQL 5.7.32要求serverTimezone=GMT%2B8（URL编码后）。
    解决：修改metrics-service/src/main/resources/application.yml：

    spring: datasource: url: jdbc:mysql://mysql-host:3306/metrics_db?useSSL=false&serverTimezone=GMT%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_TOKEN=abc123。验证：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.java：

    @PostMapping(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

    若在物理机部署，则改为实际IP：streamcube.host: 192.168.1.100。

    5. 二次开发实战：给“小微易贷”产品加一个“企业纳税额连续3年增长”规则

    提示：本章不讲“如何写Java”，而是聚焦“如何在不改核心代码的前提下，让新规则上线”。项目正文强调“支持行社个性化产品模型的配置”，这就是它的落地姿势。

    5.1 新增指标：从“企业纳税额”原始数据到可配置指标

    项目正文指标体系里有“行内数据：CRM数仓、ECIF”，但没提“企业纳税额”。我们要先把它变成平台可识别的指标，步骤如下：

    1. 在数仓中确认数据源
      找DBA确认企业纳税数据在dw_corp_tax表中，关键字段：corp_id（企业ID）、tax_year（年份）、tax_amount（纳税额）。项目正文“数仓每日进行指标批处理加工任务”，意味着此表每天凌晨ETL更新。

    2. 在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执行查询的契约。

    3. 重启metrics-service并验证

      curl "http://localhost:8082/api/v1/metrics/get?metricCode=corp.tax.amount.last3years&corpId=CORP789&year1=2021&year2=2022&year3=2023"

      应返回JSON数组：[125000, 138000, 152000]。

    5.2 配置衍生指标：把“连续3年增长”抽象成布尔型指标

    项目正文“衍生指标主要在基础指标的基础上进行简单的四则运算”，但“连续增长”是逻辑判断，需用Groovy脚本：

    1. 在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

    2. 在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脚本。

    3. 验证衍生指标

      curl "http://localhost:8082/api/v1/metrics/get?metricCode=corp.tax.growth.3years&corpId=CORP789&year1=2021&year2=2022&year3=2023"

      返回true或false。

    5.3 在规则包中引用新指标：修改XIAO_WEI_YI_DAI.xml

    找到rule/XIAO_WEI_YI_DAI.xml，在<rulePackage>内添加新规则：

    <rule name="企业纳税额连续3年增长校验" priority="5"> <condition> <![CDATA[ context.get("corp.tax.growth.3years") != null && ((Boolean)context.get("corp.tax.growth.3years")) ]]> </condition> <action>APPROVE</action> </rule>

    注意：priority="5"设为最高优先级，确保此规则最先执行。项目正文“快速发布，实现全流程各节点项实时监控”，就是靠这种低侵入式修改。

    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/目录结构里的契约——只要守住这个契约，你就能在业务需求爆发时，把上线周期从两周压缩到两小时。希望帮到你。

    本文还有配套的精品资源，点击获取