3. 实时数据管道：数据采集、消息队列、清洗与特征计算

实时风控系统的核心在于“快”与“准”。数据从产生到参与决策的端到端延迟通常需要控制在 100ms~500ms 以内。本章围绕实时数据管道的四个关键环节展开：数据采集（埋点/日志）、消息队列（Kafka/Pulsar）、数据清洗与标准化、实时特征计算。

3.1 数据采集：埋点与日志

数据采集是管道的起点，决定了数据的完整性、时效性与准确性。风控场景下，采集对象分为三类：

3.2 消息队列：Kafka vs Pulsar

消息队列承担着 削峰填谷、解耦 与 持久化 的核心职责。在实时风控场景中，Kafka 与 Pulsar 是两大主流选择。以下从风控视角进行对比：

3.3 数据清洗与标准化

原始数据往往包含脏数据、格式不一致、字段缺失等问题。清洗与标准化是保证特征质量的关键步骤。典型处理流程如下：

1. 去重：基于事件 ID（如 UUID）或业务主键，在窗口内（如 5 秒）去除重复消息。
2. 格式统一：将时间戳转为 UTC 毫秒数，枚举值映射为统一编码，金额统一为分（整数）。
3. 字段校验：检查必填字段（如 user_id、event_type、timestamp）是否为空，空值直接丢弃或写入死信队列。
4. 异常值过滤：识别并剔除明显异常数据（如年龄 > 150、金额为负数）。
5. 数据脱敏：对敏感字段（手机号、身份证）进行哈希或掩码处理，避免泄露。

3.4 实时特征计算

特征计算是风控管道的“大脑”，将原始事件转化为模型可用的数值特征。实时特征主要分为三类：

以下是一个基于 Flink SQL 的实时特征计算示例：

-- 计算每个用户过去 1 小时内的登录次数和不同 IP 数
CREATE TABLE login_events (
  user_id STRING,
  ip STRING,
  event_time TIMESTAMP(3),
  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'login_topic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);

SELECT
  user_id,
  COUNT(*) AS login_count_1h,
  COUNT(DISTINCT ip) AS distinct_ip_count_1h,
  TUMBLE_END(event_time, INTERVAL '1' HOUR) AS window_end
FROM login_events
GROUP BY
  user_id,
  TUMBLE(event_time, INTERVAL '1' HOUR);

3.5 管道可观测性

没有可观测性的管道如同盲飞。每个环节必须暴露以下指标：

• 吞吐量：每秒处理事件数（EPS），用于容量规划。
• 延迟：P50 / P95 / P99 延迟，识别瓶颈。
• 数据丢失率：（输入事件数 - 输出事件数）/ 输入事件数，应 < 0.01%。
• 死信队列大小：清洗失败的消息数量，持续增长说明清洗规则或上游数据有问题。
• 消费者 Lag：消息队列中未消费的消息数，Lag 突增可能预示下游处理能力不足。