3. 数据采集与预处理：实时数据源与特征工程

实时风险拦截模型的核心挑战在于“快”与“准”。数据从产生到进入模型推理，延迟必须控制在毫秒级。本章聚焦实时数据流的接入、清洗、特征构造与质量保障，构建从原始日志到可用特征的完整管道。

3.1 实时数据源：埋点、API 与消息队列

保险实时风险场景的数据源主要分为三类：客户端埋点、业务系统 API 以及消息队列中间件。下表对比了各自特点与适用场景：

下图展示了实时数据采集的典型三层架构：

3.2 数据清洗：实时流中的脏数据处理

实时数据流中常见的脏数据包括：字段缺失、格式异常、时间乱序、重复消息。清洗策略必须兼顾低延迟与准确性。

• 字段缺失处理：对关键字段（如用户ID、保单号）采用“丢弃”策略；对非关键字段（如设备型号）采用“默认值填充”。
• 格式标准化：统一时间戳为毫秒级 Unix 时间戳；金额字段统一为分（整数）存储，避免浮点精度问题。
• 去重机制：基于消息唯一 ID（如 traceId）在窗口内（通常 5 秒）做幂等去重，防止重复计算特征。
• 乱序处理：设置 watermark 延迟容忍度（如 10 秒），超出窗口的迟到数据直接丢弃或发往死信队列。

DataStream<RawEvent> cleanedStream = rawStream
    // 过滤空事件
    .filter(event -> event.getUserId() != null && event.getPolicyNo() != null)
    // 标准化时间戳
    .map(event -> {
        event.setTimestamp(convertToMillis(event.getTimestamp()));
        return event;
    })
    // 基于 traceId 去重（5秒滚动窗口）
    .keyBy(event -> event.getTraceId())
    .process(new DeduplicateFunction(5000L))
    // 处理乱序：允许 10 秒延迟
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<RawEvent>forBoundedOutOfOrderness(Duration.ofSeconds(10))
            .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
    );

3.3 特征工程基础：从原始日志到风险特征

实时特征工程的目标是在毫秒级时间内，将原始数据转化为模型可用的数值化特征。保险场景下，特征通常分为三类：

下图展示了实时特征工程的计算流程：

3.4 数据质量监控：实时管道的“体检”体系

数据质量是实时模型的命脉。必须建立覆盖全链路的监控体系，确保“脏数据不进、特征不丢、延迟不超”。

核心监控指标如下：