📚 课程目录第4章回放数据采集

4. 回放数据采集：网关日志格式设计（JSON/Protobuf）、Kafka作为消息中间件、日志采集的可靠性保证（ACK机制、重试策略）

回放数据采集是风控网关回放对比的基石。采集阶段的核心目标是将线上真实流量无损、有序、可靠地搬运到离线分析管道。本章聚焦三个关键决策：日志序列化格式（JSON vs Protobuf）、消息中间件选型（Kafka 核心配置）、以及端到端可靠性保证（ACK 与重试策略）。

4.1 网关日志格式设计：JSON vs Protobuf

网关日志需要记录请求头、请求体、响应体、时间戳、路由信息、风控标签等字段。序列化格式直接影响带宽、解析性能与可读性。

对比维度	JSON（文本）	Protobuf（二进制）
可读性	高，可直接查看	低，需 .proto 文件反序列化
序列化后体积	大（含字段名冗余）	小（字段用编号标识）
序列化/反序列化性能	中等（字符串解析开销）	高（二进制编码，零拷贝友好）
Schema 演进	灵活（新增字段不影响旧解析器）	需严格兼容规则（向前/向后兼容）
典型场景	调试、小流量、快速原型	高吞吐、低延迟、生产环境

实战建议：风控网关回放场景下，推荐使用 Protobuf 作为主序列化格式。原因：回放数据量级通常在 TB/天级别，Protobuf 可节省 60%-70% 的存储和带宽；同时 Spark 原生支持 Protobuf（通过 spark-protobuf 包），反序列化效率远高于 JSON。

以下是一个简化的 Protobuf 定义示例：

syntax = "proto3";

message GatewayLog {
  string trace_id = 1;          // 全局追踪ID
  int64 timestamp_ms = 2;       // 网关接收时间戳（毫秒）
  string source_ip = 3;         // 客户端IP
  string target_uri = 4;        // 请求URI
  bytes request_body = 5;       // 原始请求体（二进制）
  bytes response_body = 6;      // 原始响应体（二进制）
  int32 http_status = 7;        // HTTP状态码
  map<string, string> headers = 8; // 请求头KV
  string risk_tag = 9;          // 风控标签（如 "pass", "block"）
  int64 latency_ms = 10;        // 网关处理耗时
}

4.2 Kafka 作为消息中间件：核心配置与分区策略

Kafka 是回放数据采集的事实标准中间件。其高吞吐、持久化、可重放特性完美匹配网关日志的流式采集需求。

  关键配置清单（生产者端）：
  bootstrap.servers：集群地址列表，至少配置 3 个 broker 节点。
acks：可靠性级别，回放场景推荐 all（等待所有 ISR 副本确认）。
retries：重试次数，建议设置 Integer.MAX_VALUE 并配合 retry.backoff.ms。
compression.type：启用 snappy 或 lz4 压缩，降低网络带宽。
batch.size 与 linger.ms：平衡延迟与吞吐，建议 32KB + 5ms。

分区策略直接影响回放数据的有序性与并行度。网关日志通常按 trace_id 哈希分区，确保同一请求的日志进入同一分区，保证顺序消费。

// Java 生产者示例：按 trace_id 分区
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
props.put("acks", "all");
props.put("retries", Integer.MAX_VALUE);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
props.put("partitioner.class", "org.apache.kafka.clients.producer.RoundRobinPartitioner");
// 实际生产建议自定义分区器：按 trace_id 哈希

KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);

// 发送 Protobuf 序列化后的字节
GatewayLog log = buildGatewayLog(request, response);
ProducerRecord<String, byte[]> record = new ProducerRecord<>(
    "gateway-log-topic",
    log.getTraceId(),          // key 用于分区
    log.toByteArray()          // value 为 Protobuf 字节
);
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        // 记录发送失败日志，触发重试或死信队列
    }
});

4.3 日志采集的可靠性保证：ACK 机制与重试策略

回放对比的准确性高度依赖采集阶段的数据完整性。任何日志丢失都会导致回放结果偏差。以下从生产端和消费端两个维度阐述可靠性保证。

4.3.1 生产端可靠性：ACK 与重试

ACK 级别	行为描述	可靠性	延迟	回放场景推荐
`acks=0`	Producer 不等待任何确认	极低（可能丢数据）	最低	不推荐
`acks=1`	等待 Leader 副本写入确认	中等（Leader 宕机可能丢）	低	可接受（非核心日志）
`acks=all`	等待所有 ISR 副本确认	最高（不丢数据）	较高	强烈推荐

注意：仅设置 acks=all 还不够，必须配合 min.insync.replicas 参数。例如 Topic 副本因子为 3，设置 min.insync.replicas=2，确保至少 2 个副本写入成功才返回 ACK。这能防止单副本场景下的数据丢失。

重试策略配置模板：

# 生产者重试配置（server.properties 或 Java 客户端）
retries=2147483647          # 无限重试，直到成功或超时
retry.backoff.ms=100        # 重试间隔 100ms
max.in.flight.requests.per.connection=1  # 防止重试导致乱序（回放场景建议）
delivery.timeout.ms=120000  # 总超时 2 分钟，避免无限阻塞

4.3.2 消费端可靠性：手动提交与偏移量管理

Spark Streaming 消费 Kafka 时，默认使用 自动提交偏移量（enable.auto.commit=true），但这可能导致数据丢失或重复处理。回放场景要求精确一次语义（Exactly-Once），必须使用手动提交。

  Spark 消费端可靠性最佳实践：
  设置 enable.auto.commit=false，由 Spark 作业控制偏移量提交时机。
使用 createDirectStream 方式（非 Receiver-based），确保分区与 Spark 分区一一对应。
在 foreachRDD 或 stream.foreachBatch 中，先处理数据（写入 HDFS/Hive），处理成功后再手动提交偏移量。
将偏移量存储在外部系统（如 ZooKeeper、Redis 或 Kafka 自身 __consumer_offsets），便于故障恢复。

// Spark Structured Streaming 手动提交示例
val kafkaStream = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "kafka1:9092")
  .option("subscribe", "gateway-log-topic")
  .option("enable.auto.commit", "false")   // 关闭自动提交
  .option("startingOffsets", "earliest")   // 从头开始消费（回放场景）
  .load()

kafkaStream
  .writeStream
  .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    // 1. 将数据写入 HDFS（Parquet 格式）
    batchDF.write
      .mode("append")
      .parquet("/data/gateway-logs/")

    // 2. 手动提交偏移量（通过 Kafka 客户端 API）
    // 实际生产建议使用 Kafka 内置的 commitAsync 或 commitSync
  }
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .start()

4.3.3 端到端可靠性矩阵

环节	风险点	应对措施	效果
网关 → Producer	网络闪断、Producer 进程崩溃	本地日志缓存 + 异步批量发送	防止瞬时故障丢数据
Producer → Kafka	Leader 副本宕机、网络分区	acks=all + retries + 幂等 Producer	保证消息不丢失、不重复
Kafka 存储	磁盘故障、Broker 宕机	副本因子 ≥ 3 + ISR 机制	保证持久化数据高可用
Kafka → Spark	消费者宕机、偏移量提交失败	手动提交 + 偏移量外部存储	保证精确一次消费语义
Spark → HDFS	写入失败、文件损坏	事务性写入（如 Hive ACID）	保证目标存储数据完整性

实战经验：在风控网关回放项目中，我们曾因未设置 max.in.flight.requests.per.connection=1 导致重试时消息乱序，回放对比结果出现 0.3% 的偏差。排查两天才发现。建议回放场景严格保证分区内有序，宁可牺牲部分吞吐也要确保顺序。

4.4 本章小结

日志格式：选择 Protobuf 作为序列化方案，平衡体积与解析性能；定义清晰的 .proto 文件，预留扩展字段。
消息中间件：Kafka 配置 acks=all、min.insync.replicas=2、重试无限次，按 trace_id 哈希分区保证有序。
可靠性保证：生产端幂等 + 重试，消费端手动提交偏移量 + 外部存储，端到端实现至少一次语义（At-Least-Once），配合下游去重达到精确一次（Exactly-Once）。