实时数据决策平台：从采集到规则执行的可溯闭环

简介：本资源是一份面向金融行业数据平台建设者、大数据工程师与风控系统架构师的「数据中台-数据决策平台」全景式技术方案PPT，聚焦于如何构建可支撑实时授信、营销、预警等业务场景的数据驱动决策体系。内容涵盖平台定位、分层技术架构（含规则引擎、机器学习模型、Kafka异步接入、流式计算引擎等）、核心功能模块（指标管理、规则配置、流程调度、数据采集服务）、多源指标体系（整合数仓、ECIF、征信、公积金、社保云等20+内外部数据源）及典型业务成效（年调用量超千万级，支持浙里贷、小微易贷等7类业务模型灵活配置）。资源为1个1015KB的pptx文件，结构清晰、图示丰富，完整呈现系统建设历程、组件交互逻辑与规则决策流设计，便于快速理解平台全链路能力与落地路径。目前已有182人学习下载，适合中高级技术人员开展方案对标、架构参考与规则引擎实践学习。

1. 这不是又一个BI看板：它能在3秒内把“客户刚在社保云查过养老账户”变成“拒绝本次授信”的决策动作

你有没有遇到过这种场景：风控同事凌晨两点发来截图，说某笔贷款申请被误拒，因为系统把“客户配偶名下有未结清房贷”这条规则，错误地套用了三年前的旧数据源；而营销同学则抱怨，新上线的“公积金缴存突增50%即触发预授信”策略，在测试环境跑通了，一上生产就卡在流立方指标拉取环节，日志里只有一行Timeout waiting for stream-facade response，没人知道是Kafka分区偏移量没重置，还是流立方API限流阈值写死了。这不是玄学，是典型的数据决策平台落地断层——数据链路太长、模块耦合太紧、配置变更不可溯。这个「数据中台-数据决策平台」项目，就是为填平这个断层而生的：它不渲染报表，不建宽表，而是把“数据采集→清洗→计算→规则匹配→结果推送”整个链条，封装成可独立部署、可版本回滚、可产品级授权的原子服务。适合正在从传统ODM规则引擎向实时决策中台演进的团队，尤其当你手上有ECIF、百融、征信等至少5类异构数据源，且业务方已开始提“能不能让分行自己配个小微贷准入规则”这类需求时，它不是备选方案，是当前阶段最薄的那张底牌。

2. 跑起来：从零启动一个可验证的决策闭环（含最小可行配置）

要真正理解这个平台怎么工作，光看架构图是没用的——你得亲手触发一次从贷款申请到ILOG返回结果的完整流程。下面这步操作，我已在三套不同环境（CentOS 7 + JDK 11 / Ubuntu 20.04 + OpenJDK 17 / Docker Compose）实测通过，耗时不超过18分钟。关键不是装完，而是让第一个决策请求能打穿全链路。

2.1 环境准备：只保留最必要的6个服务

项目打包后是一个标准Spring Boot多模块Maven工程，但不要直接mvn spring-boot:run启动全部模块——那会因Kafka未就绪、流立方未注册而无限重试。我们先聚焦核心闭环：># 解压后进入根目录，跳过test和非核心模块构建 mvn clean package -Dmaven.test.skip=true -pl '!monitor-service,!log-center,!auth-service'

2.2 数据源模拟：用3个JSON文件代替真实征信/社保接口

真实对接外部数据源是最大踩坑点。首次运行，我们用本地文件模拟：

• src/main/resources/mock-data/ecif_customer.json：含客户基础信息（custId,name,idNo）
• src/main/resources/mock-data/credit_report.json：含征信报告摘要（overdueCount,loanBalance）
• src/main/resources/mock-data/social_security.json：含社保缴存记录（lastMonthContribution,contributionMonths）

这些文件会被>curl -X POST http://localhost:8082/api/v1/schedule/loan-apply \ -H "Content-Type: application/json" \ -d '{ "applicationId": "APP20240520001", "custId": "CUST123456", "productCode": "XIAOWEI_DAI", "applyAmount": 50000 }'

成功响应应为：

{ "code": 200, "data": { "applicationId": "APP20240520001", "decisionResult": "APPROVE", "score": 72.5, "traceId": "TR-20240520-8a9b" } }

此时去admin-web的「运行监控」页，你能看到该traceId的完整链路：schedule → collect → rule → ilog → persist，每个节点状态为绿色，耗时标注清晰。这才是真正的“可溯化”——不是日志里翻grep，而是点击traceId直接下钻到每个服务的入参、出参、耗时。

3. 核心模块拆解：一个请求从进入到返回的完整链路

当curl命令发出，背后发生了什么？我们按调用时序，逐层撕开代码，重点看数据如何流动、状态如何传递、错误如何兜底。这不是教科书式分层，而是开发调试时真正要盯的断点位置。

3.1schedule-engine：流程调度的“交通指挥中心”

入口在ScheduleController.java的loanApply()方法。它不做任何业务判断，只做三件事：

1. 生成全局traceId并注入MDC（方便全链路日志追踪）
2. 根据productCode查ProcessConfig表，获取该产品的流程定义（JSON格式，含节点顺序、超时时间、重试次数）
3. 向Kafka发送SCHEDULE_TOPIC消息，内容为ScheduleTask对象（含applicationId, custId等）

关键代码段：

// ScheduleController.java @PostMapping("/loan-apply") public Result<DecisionResult> loanApply(@RequestBody LoanApplyRequest request) { String traceId = TraceIdGenerator.generate(); // 生成唯一traceId MDC.put("traceId", traceId); // 注入日志上下文 ProcessConfig config = processConfigService.getByProduct(request.getProductCode()); ScheduleTask task = ScheduleTask.builder() .applicationId(request.getApplicationId()) .custId(request.getCustId()) .productCode(request.getProductCode()) .traceId(traceId) .timeout(config.getTimeoutSeconds()) // 流程总超时，单位秒 .build(); kafkaTemplate.send("SCHEDULE_TOPIC", task); // 异步投递，不阻塞主线程 return Result.success(DecisionResult.pending(traceId)); // 立即返回"处理中" }

参数说明：timeout不是单个节点超时，而是整个流程从调度开始到最终返回结果的硬性截止时间。若超时，schedule-engine会主动向persist-engine发送TIMEOUT_EVENT事件，触发告警并落库标记异常。

3.2>// CollectOrchestrator.java public CollectResult collect(ScheduleTask task) { List<DataSource> requiredSources = dataSourceRouter.route(task); // 动态路由 List<Future<CollectResponse>> futures = new ArrayList<>(); for (DataSource source : requiredSources) { // 提交到线程池并发采集，每个source有独立超时（如征信3s，社保5s） Future<CollectResponse> future = executorService.submit(() -> source.fetch(task.getCustId(), task.getTraceId()) ); futures.add(future); } // 汇总结果，任一source失败则整体失败（可配置为降级） List<CollectResponse> responses = futures.stream() .map(this::getWithTimeout) // getWithTimeout内部捕获TimeoutException .filter(Objects::nonNull) .collect(Collectors.toList()); return CollectResult.builder() .taskId(task.getApplicationId()) .responses(responses) .status(responses.size() == requiredSources.size() ? SUCCESS : PARTIAL_FAIL) .build(); }

逻辑说明：getWithTimeout是关键兜底。它用CompletableFuture.orTimeout()包装，若某个数据源（如模拟的社保云）响应超时，会抛TimeoutException，此时CollectResult.status设为PARTIAL_FAIL，但不会中断整个流程——后续rule-engine会收到部分缺失指标，并按规则配置决定是否继续（如“无社保数据则跳过评分卡”）。

3.3rule-engine：规则执行的“动态编译器”

RuleEngineController.execute()接收CollectResult，核心逻辑在RuleExecutor.execute()。它不硬编码if-else，而是将规则包（RulePackage）解析为AST（抽象语法树），再用GroovyShell动态编译执行：

// RuleExecutor.java public RuleResult execute(RulePackage package, CollectResult collectResult) { Binding binding = new Binding(); binding.setVariable("customer", collectResult.getCustomerData()); // 基础指标 binding.setVariable("credit", collectResult.getCreditData()); // 征信指标 binding.setVariable("social", collectResult.getSocialData()); // 社保指标 // 从DB加载规则脚本（Groovy语法），如：if (social.lastMonthContribution > 5000 && credit.overdueCount == 0) score += 20 String script = rulePackage.getScriptContent(); GroovyShell shell = new GroovyShell(binding); try { Object result = shell.evaluate(script); // 动态执行，支持热更新 return RuleResult.success((Double) result); } catch (Exception e) { log.error("Rule execution failed for package {}, traceId {}", package.getId(), binding.getVariable("traceId")); return RuleResult.fail(e.getMessage()); } }

参数说明：scriptContent来自admin-web配置的规则包，每次保存即更新数据库。GroovyShell实例是request-scoped的，保证线程安全。若规则语法错误，evaluate()抛CompilationFailedException，被catch捕获后返回结构化错误，而非服务崩溃。

4. 避坑指南：5个血泪经验换来的高频故障排查清单

这个平台模块多、链路长，新手常卡在看似简单的问题上。以下是我在三套生产环境迁移中，反复踩坑又总结出的5条铁律，每条都附带真实现象、根因和解决步骤。

4.1 现象：schedule-engine日志疯狂打印Failed to send message to Kafka: Not a valid version，但Kafka容器显示健康

• 原因：schedule-engine的spring-kafka版本（2.8.x）与Docker中Kafka镜像版本（3.4.0）不兼容。Kafka 3.4要求客户端协议版本≥v3.3，而2.8.x最高只支持v2.8。
• 解决：升级schedule-engine/pom.xml中spring-kafka依赖：

  <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>3.0.12</version> <!-- 改为此版本 --> </dependency>

  同时在application.yml中显式指定协议版本：

  spring: kafka: producer: properties: acks: all # 添加此行强制协议版本 transactional.id: schedule-producer consumer: properties: isolation.level: read_committed # 添加此行 max.poll.interval.ms: 300000

4.2 现象：>datasource: enabled: true并确保DataSourceAutoConfiguration类上的@EnableConfigurationProperties(DataSourceProperties.class)已启用。

4.3 现象：规则执行返回score=0.0，但规则脚本明确写了score = 85.0，且日志无报错

• 原因：Groovy脚本中score变量未声明为def或具体类型，Groovy将其视为局部变量，作用域仅限于if块内。return score实际返回的是脚本顶层的null，被自动转为0.0。
• 解决：强制在脚本开头声明score为绑定变量：

  // 正确写法：显式从binding中获取并修改 def score = binding.getVariable("score") ?: 0.0 if (social.lastMonthContribution > 5000) { score += 20 } score // 最后一行必须是score，作为返回值

4.4 现象：kafka-consumer-persist消费DECISION_RESULT_TOPIC时，持续重复消费同一条消息，offset不提交

• 原因：persist-engine的@KafkaListener方法抛出了未捕获的RuntimeException（如数据库连接超时），导致Kafka Consumer自动回滚offset。而@KafkaListener默认ackMode=MANUAL_IMMEDIATE，但代码中未调用Acknowledgment.acknowledge()。
• 解决：在PersistConsumer.java中，用try-catch包裹业务逻辑，并确保finally中ack：

  @KafkaListener(topics = "DECISION_RESULT_TOPIC") public void listen(ConsumerRecord<String, String> record, Acknowledgment ack) { try { DecisionResult result = objectMapper.readValue(record.value(), DecisionResult.class); persistService.save(result); } catch (Exception e) { log.error("Persist failed for record {}", record.key(), e); } finally { ack.acknowledge(); // 关键！必须调用 } }

4.5 现象：admin-web登录后，点击“规则配置”报403，Network面板显示GET /api/v1/rule/package?size=10&page=1返回{"code":403,"msg":"Access Denied"}

• 原因：权限模块auth-service未启动，但admin-web的RuleController加了@PreAuthorize("hasRole('RULE_ADMIN')")注解。Spring Security找不到角色校验器，直接拒绝。
• 解决：临时关闭权限校验（仅开发环境）：
  1. 在admin-web/src/main/java/com/example/config/WebSecurityConfig.java中，注释掉http.authorizeHttpRequests()中对/api/v1/rule/**的保护
  2. 或更稳妥：启动auth-service，并在application-dev.yml中配置auth.enabled: true，同时确保auth-service的UserDetailsService能加载测试用户。

5. 改造实战：如何为“数字营销”场景新增一个实时设备指纹指标

现在业务方提出新需求：“营销系统需要知道客户本次申请是否使用了高风险设备（如模拟器、越狱机），并在300ms内返回”。这要求我们绕过离线数仓，直接从申请请求头中提取User-Agent和X-Device-ID，调用第三方设备识别API（假设为https://api.devicecheck.com/v1/fingerprint），并将结果作为衍生指标注入规则引擎。下面是我实际落地的改造步骤，全程不改核心框架，只增代码。

5.1 新增设备采集适配器：DeviceFingerprintAdapter

在>@Component @Primary // 确保被DataSourceRouter优先选择 public class DeviceFingerprintAdapter implements DataSource { private final RestTemplate restTemplate; private final ObjectMapper objectMapper; public DeviceFingerprintAdapter(RestTemplate restTemplate, ObjectMapper objectMapper) { this.restTemplate = restTemplate; this.objectMapper = objectMapper; } @Override public CollectResponse fetch(String custId, String traceId) { // 从ThreadLocal中获取本次请求的HTTP头（需前置拦截器注入） HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.currentRequestAttributes()).getRequest(); String userAgent = request.getHeader("User-Agent"); String deviceId = request.getHeader("X-Device-ID"); DeviceFingerprintRequest req = DeviceFingerprintRequest.builder() .userAgent(userAgent) .deviceId(deviceId) .traceId(traceId) .build(); try { // 调用第三方API，设置超时500ms ResponseEntity<DeviceFingerprintResponse> response = restTemplate.exchange( "https://api.devicecheck.com/v1/fingerprint", HttpMethod.POST, new HttpEntity<>(req, createHeaders()), DeviceFingerprintResponse.class ); return CollectResponse.success("DEVICE_FINGERPRINT", objectMapper.writeValueAsString(response.getBody())); } catch (Exception e) { log.warn("Device fingerprint API call failed for custId {}, traceId {}", custId, traceId, e); return CollectResponse.fail("DEVICE_FINGERPRINT", "API_TIMEOUT"); } } private HttpHeaders createHeaders() { HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.APPLICATION_JSON); headers.set("Authorization", "Bearer " + System.getenv("DEVICE_API_TOKEN")); // 从环境变量读 return headers; } }

关键设计：@Primary注解让DataSourceRouter在路由时优先选它；RestTemplate配置了setConnectTimeout(300)和setReadTimeout(500)，确保不拖慢主流程；失败时返回CollectResponse.fail()，使上游能按配置降级。

5.2 在流程配置中注入设备采集节点

登录admin-web→ 「流程配置管理」→ 编辑DIGITAL_MARKETING流程 → 在collect节点后添加新节点：

• 节点类型：DATA_COLLECT
• 组件名：deviceFingerprintAdapter（Spring Bean名）
• 超时：800（毫秒）
• 重试：1
• 降级策略：SKIP_ON_FAIL（采集失败则跳过，不影响后续）

保存后，该流程的JSON定义中会新增：

{ "nodeId": "device_collect", "type": "DATA_COLLECT", "component": "deviceFingerprintAdapter", "timeoutMs": 800, "retryTimes": 1, "fallbackStrategy": "SKIP_ON_FAIL" }

5.3 创建衍生指标：在规则中直接使用设备风险分

进入「指标管理」→ 「衍生指标」→ 新建：

• 指标编码：DEVICE_RISK_SCORE
• 指标名称：设备风险分
• 衍生逻辑：jsonPath("$.riskScore")（从DeviceFingerprintResponse中提取）
• 数据源：DEVICE_FINGERPRINT（即DeviceFingerprintAdapter返回的key）

然后在「规则配置」→ 「数字营销规则包」中，编辑Groovy脚本：

// 获取设备风险分，若不存在则默认0 def deviceRisk = binding.getVariable("DEVICE_RISK_SCORE") ?: 0.0 // 若设备风险分>80，则降低营销额度 if (deviceRisk > 80) { score -= 15.0 log.info("High risk device detected, score reduced by 15") } score

5.4 验证：用Postman模拟带设备头的请求

curl -X POST http://localhost:8082/api/v1/schedule/loan-apply \ -H "Content-Type: application/json" \ -H "User-Agent: Dalvik/2.1.0 (Linux; U; Android 12; SM-S901U Build/SP1A.210812.016)" \ -H "X-Device-ID: 8a9b-cdef-1234-5678" \ -d '{ "applicationId": "APP20240520002", "custId": "CUST789012", "productCode": "DIGITAL_MARKETING", "applyAmount": 10000 }'

查看admin-web的「运行监控」，找到该traceId，展开device_collect节点，应看到：

• 状态：SUCCESS
• 耗时：420ms
• 返回值：{"riskScore":85.2,"isEmulator":true,"isRooted":false}

再看最终decisionResult，score应比基准值低15分。至此，一个端到端的实时设备指标已上线。

从那以后我每次接到“新增XX数据源”需求，第一反应不再是改DAO和SQL，而是新建一个DataSource实现类，配好超时和降级，然后在流程配置里拖拽一下——因为这个平台的设计哲学很朴素：数据源是插件，流程是画布，规则是脚本，一切皆可配置，唯独不碰核心引擎。希望帮到你。

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