4. Kafka 集成与数据接入

在实时模型漂移巡查场景中，Kafka 是流数据的核心传输层。模型特征数据、预测结果、标签回传均通过 Kafka 主题流转。本章聚焦 Kafka 生产者/消费者配置、Spark 读取 Kafka 流、Schema Registry 与 Avro 序列化，以及 Exactly-Once 语义保证，确保数据接入环节高可靠、低延迟、不丢不重。

4.1 Kafka 生产者与消费者配置

生产者和消费者的参数配置直接影响数据吞吐量与端到端延迟。以下为模型漂移场景下的推荐配置清单。

4.2 Spark 读取 Kafka 流

Spark Structured Streaming 通过 readStream 接口集成 Kafka，支持微批次与连续处理两种模式。模型漂移巡查通常采用微批次模式，兼顾吞吐与 Exactly-Once 语义。

以下为 Spark 读取 Kafka 流的标准代码模板：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.Trigger

val spark = SparkSession.builder()
  .appName("ModelDriftKafkaIngestion")
  .config("spark.sql.streaming.schemaInference", "true")
  .getOrCreate()

val kafkaStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
  .option("subscribe", "model-features,model-predictions")
  .option("startingOffsets", "earliest")
  .option("failOnDataLoss", "false")
  .option("maxOffsetsPerTrigger", "10000")
  .load()

// 解析 Kafka 消息 key 与 value（二进制格式）
val parsedStream = kafkaStream
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "topic", "partition", "offset", "timestamp")

4.3 Schema Registry 与 Avro 序列化

模型特征数据通常包含数十到数百个字段，使用 Avro 序列化可以大幅压缩数据体积，并通过 Schema Registry 实现 schema 版本管理。当模型特征结构发生变化（如新增特征列），Schema Registry 能向后兼容，避免消费端解析失败。

Avro 序列化在模型漂移场景中的核心优势：

• 紧凑二进制格式：比 JSON 小 60%-80%，降低 Kafka 存储与网络开销。
• Schema 演进：支持添加/删除字段，兼容性策略可配置（BACKWARD / FORWARD / FULL）。
• 强类型校验：生产端与消费端 schema 必须匹配，提前发现数据质量问题。

以下为 Avro schema 定义示例（模型特征数据）：

{
  "namespace": "com.mldrift.features",
  "type": "record",
  "name": "ModelFeatureRecord",
  "fields": [
    {"name": "model_id", "type": "string"},
    {"name": "feature_vector", "type": {"type": "array", "items": "double"}},
    {"name": "feature_names", "type": {"type": "array", "items": "string"}},
    {"name": "inference_timestamp", "type": "long"},
    {"name": "data_partition", "type": "string", "default": "online"}
  ]
}

Spark 集成 Schema Registry 需要引入 Confluent Avro 格式依赖，并在读取 Kafka 流时指定反序列化器：

import org.apache.spark.sql.avro._

val avroStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092")
  .option("subscribe", "model-features-avro")
  .load()
  .select(
    from_avro(col("value"), "ModelFeatureRecord", 
      Map("schema.registry.url" -> "http://schema-registry:8081")).as("features")
  )

4.4 Exactly-Once 语义保证

模型漂移巡查要求数据不丢不重，否则漂移指标计算会产生偏差。Spark Structured Streaming 结合 Kafka 的幂等生产者与事务性 API，可以实现端到端的 Exactly-Once 语义。

实现 Exactly-Once 的三个关键层级：

Spark 结构化流中启用 Exactly-Once 的核心配置：

val query = parsedStream
  .writeStream
  .outputMode("append")
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092")
  .option("topic", "drift-metrics-output")
  .option("checkpointLocation", "/data/checkpoints/drift-metrics")
  .option("kafka.enable.idempotence", "true")
  .option("kafka.transactional.id", "drift-metrics-txn-001")
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .start()