2. 实时数据源接入架构：消息队列选型与可靠性保障

实时风控模型训练集构建的第一步，是将分散在各业务系统、服务器、网关的日志与事件数据，以毫秒级延迟、零丢失的方式接入中央处理管道。本章聚焦消息队列选型（Kafka vs. Pulsar）、日志采集工具（Flume vs. Filebeat）以及端到端的可靠性保障机制。

2.1 消息队列选型：Kafka 与 Pulsar 对比

消息队列是实时数据接入的“中枢神经”。在风控场景下，选型需重点评估吞吐量、端到端延迟、数据持久化、分区扩展性以及多租户隔离能力。下表从风控工程视角对 Apache Kafka 和 Apache Pulsar 进行对比：

2.2 实时日志采集：Flume 与 Filebeat 对比

日志采集 Agent 负责从应用服务器、容器、网关实时读取日志文件或监听网络端口，并将数据发送至消息队列。Flume 和 Filebeat 是两种主流选择：

2.3 数据接入可靠性保障机制

风控模型训练集对数据完整性要求极高——丢失一条交易日志可能导致模型漏判欺诈。以下从生产者、消息队列、消费者三个层面构建可靠性保障：

1. 生产者端（Agent → MQ）：
   • 启用 acks=all（Kafka）或 Ensemble Size=3, Write Quorum=2（Pulsar），确保消息写入多数副本后才返回成功。
   • 本地磁盘缓存（Filebeat Registry / Flume File Channel）记录已发送偏移量，进程重启后自动续传。
   • 设置 重试次数（retries=5）与 指数退避，避免网络抖动导致瞬时失败。
2. 消息队列端（Broker 集群）：
   • 副本因子（Replication Factor）≥ 3，且最小 ISR（In-Sync Replicas）≥ 2。
   • 启用 unclean.leader.election=false（Kafka）或 AutoRecovery（Pulsar），防止落后副本被选为 Leader 导致数据丢失。
   • 磁盘容量监控与日志保留策略（Retention Policy）配合，避免磁盘写满导致写入失败。
3. 消费者端（MQ → 训练集存储）：
   • 使用 手动提交偏移量（Manual Commit），仅在数据成功写入下游存储（HDFS/对象存储/特征库）后才提交。
   • 实现 幂等消费：为每条消息生成唯一 ID（如 UUID 或业务主键），下游去重。
   • 设置 消费超时与重试队列：消费失败的消息进入死信队列（DLQ），人工或自动补偿。

2.4 实时数据接入架构总览

下图展示了一个典型的风控实时数据接入架构，涵盖 Filebeat 采集、Kafka/Pulsar 消息队列、以及端到端的可靠性保障节点：

2.5 关键配置示例

以下给出 Filebeat 采集日志并发送至 Kafka 的核心配置（filebeat.yml），以及 Kafka 生产者端可靠性参数：

# filebeat.yml —— 采集层配置
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/app/*.log
  fields:
    app_id: "payment-service"
    env: "prod"
  fields_under_root: true

output.kafka:
  hosts: ["kafka-broker1:9092", "kafka-broker2:9092", "kafka-broker3:9092"]
  topic: "risk-raw-log"
  required_acks: -1          # 等价于 acks=all
  compression: gzip
  max_message_bytes: 1048576 # 1MB
  worker: 4
  bulk_max_size: 2048

# Registry 文件记录偏移量，保证重启续传
filebeat.registry.path: /var/lib/filebeat/registry

# Kafka 生产者端可靠性参数（Java 示例）
Properties props = new Properties();
props.put("bootstrap.servers", "kafka-broker1:9092,kafka-broker2:9092");
props.put("acks", "all");                          // 等待所有副本确认
props.put("retries", 5);                           // 重试次数
props.put("retry.backoff.ms", 500);                // 重试间隔
props.put("enable.idempotence", true);             // 幂等生产者
props.put("max.in.flight.requests.per.connection", 1); // 保证顺序
props.put("linger.ms", 10);                        // 批量发送延迟
props.put("batch.size", 65536);                    // 64KB 批次