1. 课程导论与实验环境准备

1.1 贷后预警业务背景

在金融信贷全生命周期中，贷后管理是风险控制的核心环节。传统贷后监控依赖人工报表与离线跑批，存在时效性差、规则僵化、无法应对突发风险等痛点。随着实时计算技术的发展，基于 Apache Flink 构建的实时贷后预警系统能够实现毫秒级风险识别，覆盖以下典型场景：

• 资金流向异常：借款人放款后资金迅速回流至关联账户或敏感行业。
• 多头借贷预警：同一客户在短时间内向多家机构发起借款申请。
• 还款行为突变：历史还款规律被打破，如提前还款、逾期频次骤增。
• 抵押物价值波动：实时接入外部估值数据，触发抵押率阈值告警。

1.2 实验目标

本课程通过一个完整的贷后预警实验项目，帮助学习者掌握以下核心能力：

维度	具体目标	对应技术栈
环境搭建	独立完成 Flink Standalone 集群、Kafka、MySQL 部署	Linux、Shell、Docker（可选）
实时接入	实现 Kafka 数据源连接与 Schema 注册	Flink Kafka Connector
规则引擎	基于 Flink CEP 与动态规则实现预警逻辑	Flink CEP、状态后端
结果存储	将预警结果写入 MySQL，并支持回溯查询	JDBC Sink、事务管理
性能评估	压测场景下吞吐量与延迟指标分析	Flink Metrics、Prometheus

1.3 实验环境架构总览

下图展示了本次实验的完整数据流与组件关系：

1.4 Flink 集群搭建（Standalone 模式）

本实验采用 Flink Standalone 集群 部署，适合学习与开发环境。以下为搭建步骤：

1.4.1 环境要求

组件	版本要求	说明
JDK	1.8 或 11	Flink 1.14+ 推荐 JDK 11
操作系统	Linux / macOS / WSL2	生产环境建议 CentOS 7+
SSH 免密登录	主节点到所有 Worker	用于远程启动 TaskManager
网络	节点间 6123/8081 端口互通	JobManager 与 Web UI 端口

1.4.2 安装与配置

1. 下载解压

wget https://archive.apache.org/dist/flink/flink-1.14.6/flink-1.14.6-bin-scala_2.12.tgz
tar -xzf flink-1.14.6-bin-scala_2.12.tgz -C /opt/
cd /opt/flink-1.14.6

2. 配置 conf/flink-conf.yaml

jobmanager.rpc.address: node01
jobmanager.rpc.port: 6123
jobmanager.memory.process.size: 2048m
taskmanager.memory.process.size: 2048m
taskmanager.numberOfTaskSlots: 4
parallelism.default: 2

3. 配置 workers 文件

# conf/workers
node01
node02
node03

4. 启动集群

# 在主节点执行
./bin/start-cluster.sh

# 验证进程
jps  # 应看到 StandaloneSessionClusterEntrypoint 和 TaskManagerRunner

5. 访问 Web UI

浏览器打开 http://node01:8081，确认集群状态为 "Running"

1.5 Kafka 环境部署

Kafka 作为实时数据管道，负责接收模拟的贷款事件并传递给 Flink。本实验使用 Kafka 2.8.1（内置 ZooKeeper 模式）。

1.5.1 安装步骤

wget https://archive.apache.org/dist/kafka/2.8.1/kafka_2.12-2.8.1.tgz
tar -xzf kafka_2.12-2.8.1.tgz -C /opt/
cd /opt/kafka_2.12-2.8.1

1.5.2 启动服务

# 启动 ZooKeeper（Kafka 内置）
bin/zookeeper-server-start.sh config/zookeeper.properties &

# 启动 Kafka Broker
bin/kafka-server-start.sh config/server.properties &

# 创建实验 Topic
bin/kafka-topics.sh --create --topic loan_events \
  --bootstrap-server localhost:9092 \
  --partitions 3 --replication-factor 1

1.5.3 验证连通性

# 生产者测试
bin/kafka-console-producer.sh --topic loan_events --bootstrap-server localhost:9092

# 消费者测试（另开终端）
bin/kafka-console-consumer.sh --topic loan_events --bootstrap-server localhost:9092 --from-beginning

1.6 MySQL 数据源准备

MySQL 用于存储预警规则配置与预警结果记录。本实验使用 MySQL 8.0，创建专用数据库与表结构。

1.6.1 建库与用户

CREATE DATABASE IF NOT EXISTS loan_early_warning
  DEFAULT CHARACTER SET utf8mb4
  DEFAULT COLLATE utf8mb4_unicode_ci;

CREATE USER 'warning_user'@'%' IDENTIFIED BY 'Warning@2024';
GRANT ALL PRIVILEGES ON loan_early_warning.* TO 'warning_user'@'%';
FLUSH PRIVILEGES;

1.6.2 预警结果表

USE loan_early_warning;

CREATE TABLE t_warning_result (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  loan_id       VARCHAR(64)   NOT NULL COMMENT '贷款编号',
  customer_id   VARCHAR(32)   NOT NULL COMMENT '客户编号',
  rule_id       INT           NOT NULL COMMENT '触发规则ID',
  rule_name     VARCHAR(128)  NOT NULL COMMENT '规则名称',
  warning_level TINYINT       NOT NULL DEFAULT 1 COMMENT '预警等级:1低 2中 3高',
  event_time    DATETIME(3)   NOT NULL COMMENT '事件发生时间',
  warning_time  DATETIME(3)   NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '预警生成时间',
  detail_json   JSON          DEFAULT NULL COMMENT '事件详情JSON',
  INDEX idx_loan_id (loan_id),
  INDEX idx_customer_id (customer_id),
  INDEX idx_warning_time (warning_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='贷后预警结果表';

1.6.3 规则配置表（可选）

CREATE TABLE t_rule_config (
  rule_id       INT AUTO_INCREMENT PRIMARY KEY,
  rule_name     VARCHAR(128)  NOT NULL,
  rule_expression VARCHAR(512) NOT NULL COMMENT '规则表达式或CEP模式',
  rule_status   TINYINT       NOT NULL DEFAULT 1 COMMENT '1启用 0停用',
  create_time   DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP,
  update_time   DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='预警规则配置表';

1.7 环境验证清单

完成上述部署后，请逐项确认以下检查点：

检查项	验证方法	预期结果
Flink 集群状态	访问 http://localhost:8081	TaskManager 在线，Slot 可用
Kafka 连通性	生产/消费一条测试消息	消息正常收发
MySQL 连接	使用 JDBC 客户端测试	可执行 SELECT 1
端口可达	telnet 各组件端口	6123/8081/9092/3306 均开放