3、流式机器学习管道概述：离线训练 vs 在线推理、流式推理架构、模型版本管理

在构建流式机器学习管道时，需要明确区分离线训练与在线推理的职责边界。离线训练侧重于利用历史批量数据构建高精度模型，而在线推理则要求模型在毫秒级延迟内对实时数据做出预测。本章将系统阐述这两种模式的差异、流式推理的典型架构以及模型版本管理的核心策略。

3.1 离线训练 vs 在线推理

离线训练与在线推理在数据来源、计算模式、延迟要求、资源消耗等方面存在本质区别。下表从多个维度对比两者的关键差异：

3.2 流式推理架构

流式推理架构的核心是将训练好的模型部署到流处理引擎中，对实时数据流进行逐条或微批预测。典型的流式推理架构包含数据接入、特征工程、模型推理、结果输出四个阶段。

下图展示了基于 Spark Structured Streaming 的流式推理架构：

3.3 模型版本管理

模型版本管理是流式机器学习管道中衔接训练与推理的关键环节。没有完善的版本管理，模型漂移监控将无法准确追溯模型行为变化，回滚操作也会变得混乱。

下表列出了模型版本管理中的核心概念与推荐实践：

下图展示了模型版本管理的生命周期流程：

3.4 流式推理代码示例

以下代码展示了如何在 Spark Structured Streaming 中加载 MLflow 模型并进行实时推理：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.mlflow.spark.autologging.MlflowAutologEventPublisher
import org.mlflow.tracking.MlflowClient

// 初始化 Spark Session
val spark = SparkSession.builder()
  .appName("StreamingInference")
  .config("spark.sql.streaming.schemaInference", "true")
  .getOrCreate()

// 从 MLflow Model Registry 加载模型（指定版本）
val modelUri = "models:/fraud_detection_model/Production"
val model = spark.read.mlflow.load(modelUri)

// 定义流式数据源
val streamDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "transactions")
  .load()
  .selectExpr("CAST(value AS STRING) as json")
  .select(from_json(col("json"), schema).as("data"))
  .select("data.*")

// 特征工程与推理
val predictions = model.transform(streamDF)
  .withColumn("prediction_timestamp", current_timestamp())
  .withColumn("model_version", lit("v2.1.0"))

// 输出结果到 Kafka
val query = predictions.selectExpr("to_json(struct(*)) AS value")
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("topic", "predictions")
  .option("checkpointLocation", "/tmp/checkpoint")
  .outputMode("append")
  .start()

query.awaitTermination()