2. Spark结构化流基础回顾：DataStreamReader、DataStreamWriter、Trigger、Checkpoint、OutputMode

在构建模型漂移监控管道之前，必须扎实掌握Spark结构化流（Structured Streaming）的核心抽象。结构化流将数据流视为一张“无界表”（Unbounded Table），持续追加新数据，并支持基于事件时间的窗口聚合与状态管理。本章聚焦五个基础构件：DataStreamReader（数据接入）、DataStreamWriter（数据输出）、Trigger（触发策略）、Checkpoint（容错机制）和OutputMode（输出语义）。

2.1 DataStreamReader：从源头读取流数据

DataStreamReader 是结构化流的入口，负责从各种数据源（Kafka、文件、Socket、Rate等）创建 DataFrame。它通过链式方法设置格式、选项和模式，最终调用 load() 返回一个流式 DataFrame。

// 示例：从Kafka读取模型推理日志
import org.apache.spark.sql.types._

val schema = StructType(Array(
  StructField("event_time", TimestampType),
  StructField("model_id", StringType),
  StructField("feature_vector", ArrayType(DoubleType)),
  StructField("prediction", DoubleType),
  StructField("ground_truth", DoubleType)
))

val kafkaDF = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
  .option("subscribe", "model-inference-log")
  .option("startingOffsets", "earliest")
  .option("failOnDataLoss", "false")
  .load()
  .selectExpr("CAST(value AS STRING) as json_str")
  .select(from_json(col("json_str"), schema).as("data"))
  .select("data.*")

2.2 DataStreamWriter：将结果写出到外部系统

DataStreamWriter 是结构化流的输出终端，负责将流式查询的结果以指定模式写出到Kafka、文件、控制台或自定义接收器。它通过 writeStream 方法从DataFrame创建，并链式配置输出格式、输出模式、触发器、检查点位置等。

// 示例：将漂移检测结果写出到Kafka告警topic
val driftAlertDF = ... // 经过漂移检测逻辑后的DataFrame

val query = driftAlertDF
  .selectExpr("to_json(struct(*)) AS value")
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092")
  .option("topic", "model-drift-alert")
  .option("checkpointLocation", "/data/checkpoints/drift-detector")
  .outputMode("append")
  .trigger(Trigger.ProcessingTime("30 seconds"))
  .start()

query.awaitTermination()

2.3 Trigger：控制数据处理的节奏

Trigger决定了流式查询何时检查新数据并启动一次微批次处理。结构化流支持三种触发策略，适用于不同的延迟与吞吐需求。

// 三种Trigger的代码示例
import org.apache.spark.sql.streaming.Trigger

// 1. ProcessingTime：每30秒触发一次
df.writeStream.trigger(Trigger.ProcessingTime("30 seconds"))

// 2. Once：只处理一次当前所有可用数据，然后停止
df.writeStream.trigger(Trigger.Once())

// 3. Continuous：实验性，每1毫秒检查一次（需设置checkpoint）
df.writeStream.trigger(Trigger.Continuous("1 second"))

2.4 Checkpoint：容错与状态恢复的基石

Checkpoint目录是结构化流实现“精确一次”（Exactly-Once）语义的核心。它持久化三部分信息：偏移量（已处理数据的进度）、状态数据（如聚合中间结果）、提交日志（已完成批次）。当作业失败重启时，Spark从Checkpoint恢复偏移量和状态，确保数据不丢不重。

// 设置Checkpoint的两种方式
// 方式1：在DataStreamWriter中指定
df.writeStream
  .option("checkpointLocation", "hdfs://namenode:8020/checkpoints/model-drift")
  .start()

// 方式2：在SparkSession级别设置默认（不推荐，因为所有查询共享）
spark.conf.set("spark.sql.streaming.checkpointLocation", "/default/checkpoint")

2.5 OutputMode：定义输出的语义边界

OutputMode决定了每次微批次触发时，哪些行会被写入外部系统。它直接影响结果表的可见性和状态管理方式。

// OutputMode 代码示例

// Append模式：仅输出新增的推理记录（无状态）
inferenceDF.writeStream
  .outputMode("append")
  .format("parquet")
  .option("path", "/data/inference-logs")
  .start()

// Complete模式：输出每个模型的聚合统计（必须有groupBy）
val aggDF = inferenceDF.groupBy("model_id").agg(avg("prediction").as("avg_pred"))
aggDF.writeStream
  .outputMode("complete")
  .format("console")
  .start()

// Update模式：输出状态发生变化的行
val stateDF = inferenceDF.groupBy("model_id").agg(collect_list("prediction").as("preds"))
stateDF.writeStream
  .outputMode("update")
  .format("console")
  .start()

2.6 五要素协同工作流

将上述五个构件串联起来，形成一个完整的结构化流作业生命周期：

1. 定义Schema：为输入数据（如Kafka消息中的JSON/Avro）定义明确的StructType。
2. 创建DataStreamReader：配置数据源格式、连接参数、起始偏移量，调用 load() 得到流式DataFrame。
3. 应用业务逻辑：对DataFrame进行过滤、转换、聚合、漂移检测等操作。
4. 配置DataStreamWriter：选择输出格式、OutputMode、Trigger、Checkpoint路径。
5. 启动查询：调用 start() 返回 StreamingQuery 对象，可通过 awaitTermination() 保持作业运行。
6. 故障恢复：作业失败后重启，Spark自动从Checkpoint目录恢复偏移量和状态，继续处理。

// 完整示例：漂移监控作业骨架
val schema = StructType(Array(
  StructField("model_id", StringType),
  StructField("timestamp", TimestampType),
  StructField("feature_hash", StringType),
  StructField("score", DoubleType)
))

val sourceDF = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "kafka:9092")
  .option("subscribe", "model-scores")
  .option("startingOffsets", "earliest")
  .load()
  .selectExpr("CAST(value AS STRING) as json")
  .select(from_json(col("json"), schema).as("data"))
  .select("data.*")

val driftDF = sourceDF
  .groupBy("model_id")
  .agg(
    mean("score").as("avg_score"),
    stddev("score").as("std_score")
  )

driftDF.writeStream
  .outputMode("update")
  .format("console")
  .option("checkpointLocation", "/data/checkpoint/drift-demo")
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .start()
  .awaitTermination()