📚 课程目录第4章 PSI漂移告警数据采集架构

4. PSI漂移告警数据采集架构：实时流式与离线批量

在可疑交易监测场景中，PSI（群体稳定性指标）漂移告警依赖高质量、低延迟的特征数据。数据采集架构需同时支撑实时流式（秒级/分钟级特征分布监控）与离线批量（T+1全量分布基线计算）两条链路。本章详细拆解基于Kafka/Flink的实时采集与基于Hive/Spark的离线采集架构设计。

4.1 双链路采集架构总览

实时链路负责交易事件流接入、特征提取与滑动窗口PSI计算；离线链路负责历史全量数据重算、分布基线生成与模型回测。两条链路在数据存储层（HDFS/对象存储）交汇，统一供给告警引擎。

  架构核心原则：实时链路保证低延迟（P99 < 5s），离线链路保证高吞吐（单日百亿级交易量），两套管道互不阻塞，通过统一的特征元数据中心保持口径一致。

4.2 实时流式数据采集（Kafka + Flink）

实时链路以Kafka作为消息总线，Flink作为流处理引擎，完成从交易事件接入到特征分布聚合的全流程。

4.2.1 核心组件与职责

组件	职责	关键配置/参数
Kafka Source	接入交易事件流，支持多分区高吞吐	topic.replication.factor=3, min.insync.replicas=2
Flink DataStream	实时特征提取、数据清洗、类型转换	checkpoint间隔60s, exactly-once语义
滑动窗口聚合	按特征维度统计分布直方图（分钟级）	窗口大小10min, 滑动步长1min
PSI计算算子	对比当前分布与基线分布，输出PSI值	分箱数=20, 平滑因子ε=1e-6
Kafka Sink	将PSI结果写入告警Topic	acks=all, compression.type=snappy

4.2.2 Flink作业核心代码片段

// 实时PSI计算Flink作业骨架
DataStream<TransactionEvent> eventStream = env
    .addSource(new FlinkKafkaConsumer<>("txn-events", new AvroDeserializationSchema(), kafkaProps));

DataStream<FeatureDistribution> distStream = eventStream
    .keyBy(event -> event.getFeatureName())
    .window(SlidingProcessingTimeWindows.of(Time.minutes(10), Time.minutes(1)))
    .aggregate(new DistributionAggregator());

DataStream<PSIResult> psiStream = distStream
    .connect(baselineStream)  // 从状态后端或外部存储加载基线
    .process(new PSICalculator());

psiStream.addSink(new FlinkKafkaProducer<>("psi-alert", new JsonSerializationSchema(), kafkaProps));
env.execute("RealTime PSI Monitor");

生产环境避坑：实时PSI计算对状态后端压力较大，建议使用RocksDBStateBackend + 增量checkpoint；滑动窗口不宜过小（小于5min会导致频繁触发），推荐10min窗口+1min滑动步长。

4.3 离线批量数据采集（Hive + Spark）

离线链路负责T+1全量数据重算，生成特征分布基线、历史PSI趋势表以及模型回测数据集。采用Spark SQL + Hive分区表架构，支持增量与全量两种调度模式。

4.3.1 离线ETL流程

数据接入：从Kafka落盘HDFS（Parquet格式，按天分区），或直接从Hive ODS层读取。
特征清洗：过滤异常值、空值填充、类型统一，生成宽表 feature_wide_table。
分布计算：按特征+日期分组，计算等频分箱后的样本计数与占比。
基线生成：取最近N天（默认30天）的分布加权平均作为基线，存入 psi_baseline 表。
PSI计算：逐日对比当日分布与基线，输出PSI值及漂移特征列表。

4.3.2 Spark SQL示例

-- 离线PSI基线生成SQL（Spark SQL）
INSERT OVERWRITE TABLE psi_baseline PARTITION(dt = '${yesterday}')
SELECT
    feature_name,
    bucket_id,
    SUM(cnt) / SUM(SUM(cnt)) OVER (PARTITION BY feature_name) AS baseline_ratio
FROM (
    SELECT
        feature_name,
        ntile(20) OVER (PARTITION BY feature_name ORDER BY feature_value) AS bucket_id,
        COUNT(*) AS cnt
    FROM feature_wide_table
    WHERE dt BETWEEN date_sub('${yesterday}', 29) AND '${yesterday}'
    GROUP BY feature_name, feature_value
) t
GROUP BY feature_name, bucket_id;

离线基线更新策略：建议采用滑动窗口（30天）而非固定全量，避免因业务季节性波动导致基线失真；同时保留快照版本，支持PSI回测与告警回溯。

4.4 实时与离线链路对比

对比维度	实时流式（Kafka+Flink）	离线批量（Hive+Spark）
延迟	秒级～分钟级	T+1（小时级）
吞吐	单作业10万+ events/s	百亿级/天
计算语义	Exactly-once（checkpoint）	幂等写（分区覆盖）
状态管理	RocksDB状态后端	Hive元数据 + HDFS
典型产出	分钟级PSI告警事件	日级PSI趋势表、基线快照
适用场景	实时交易监控、快速漂移发现	模型回测、报表分析、基线校准

4.5 架构设计关键要点

特征元数据中心：实时与离线共用同一份特征定义（特征名、分箱数、数据类型），避免口径不一致导致PSI计算偏差。
数据质量监控：在采集链路中嵌入数据质量检查（空值率、唯一值数、类型异常），异常数据单独写入死信队列（DLQ）并触发告警。
容错与重试：Flink作业启用checkpoint + 重启策略；离线Spark任务启用shuffle服务 + 推测执行。
存储选型：实时中间结果存RocksDB或Redis；离线基线存Hive ORC格式（高压缩比）；PSI告警结果存Elasticsearch用于快速检索。

4.6 本章小结

PSI漂移告警的数据采集架构采用实时流式（Kafka+Flink）与离线批量（Hive+Spark）双轨并行模式。实时链路负责分钟级PSI计算与快速告警，离线链路负责日级基线校准与历史回测。两条链路通过统一特征存储和元数据中心保持口径一致，确保告警准确性与可回溯性。在实际部署中，需重点关注状态管理、数据质量监控以及基线更新策略，以应对金融交易场景下的高吞吐与低延迟挑战。