如何在Kafka与Spark结构化流中应用ML，避免AnalysisException异常

错误原因 #

你遇到的pyspark.sql.utils.AnalysisException是因为流式DataFrame（通过readStream创建）不能直接使用批处理API。你的代码里，randomSplit、lr.fit()、describe().show()这些都是批处理操作，而流式DataFrame的所有计算逻辑必须通过writeStream触发执行，无法像批处理DataFrame那样直接调用这类方法。

修改思路 #

Spark结构化流处理实时机器学习，通常用foreachBatch实现按批次训练/更新模型——每次有新数据批次到达时，用该批次数据训练或更新模型，同时输出结果。另外需要修正代码里的几个细节问题：

• schema字段名笔误：cel2、cel3应改为col2、col3
• col1定义为StringType，无法作为VectorAssembler的特征输入，需改为数值类型（如DoubleType）或更换特征列

修正后的代码 #

from pyspark.sql import SparkSession
from pyspark.ml.regression import LinearRegression
from pyspark.ml.feature import VectorAssembler
from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StructField, DoubleType
from pyspark.ml.evaluation import RegressionEvaluator

# 初始化Spark会话
spark = SparkSession.builder.appName('RealTimeML').getOrCreate()
spark.sparkContext.setLogLevel("WARN")

# 修正schema：修正字段名、col1改为数值类型
schema = StructType([
    StructField("Index", DoubleType(), True),
    StructField("col1", DoubleType(), True),
    StructField("col2", DoubleType(), True),
    StructField("col3", DoubleType(), True),
    StructField("col4", DoubleType(), True),
    StructField("col5", DoubleType(), True)
])

# 读取Kafka流式数据
stream_df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "topic") \
    .load() \
    .select(from_json(col("value").cast("string"), schema).alias("data")) \
    .selectExpr("data.*")

# 特征组装器：使用数值型列作为特征输入
assembler = VectorAssembler(
    inputCols=["col1", "col2", "col3"],
    outputCol="features",
    handleInvalid="keep"
)

# 全局变量存储模型，用于增量更新
global_model = None

def process_batch(df, batch_id):
    global global_model
    
    # 跳过空批次，避免无数据时报错
    if df.count() == 0:
        return
    
    # 对当前批次数据做特征转换
    df = assembler.transform(df)
    
    # 拆分当前批次的训练/测试集（仅针对该批次的批处理操作）
    splits = df.randomSplit([0.7, 0.3])
    train_df = splits[0]
    test_df = splits[1]
    
    # 首次训练或增量更新模型
    if global_model is None:
        lr = LinearRegression(featuresCol='features', labelCol='col4', maxIter=10, regParam=0.3, elasticNetParam=0.8)
        global_model = lr.fit(train_df)
        print(f"=== 批次 {batch_id}: 首次模型训练完成 ===")
    else:
        global_model = global_model.fit(train_df)
        print(f"=== 批次 {batch_id}: 模型增量更新完成 ===")
    
    # 输出模型核心指标
    print(f"系数: {global_model.coefficients}")
    print(f"截距: {global_model.intercept}")
    training_summary = global_model.summary
    print(f"训练集RMSE: {training_summary.rootMeanSquaredError:.4f}")
    print(f"训练集R2: {training_summary.r2:.4f}")
    
    # 测试集评估
    lr_predictions = global_model.transform(test_df)
    evaluator = RegressionEvaluator(predictionCol="prediction", labelCol="col4", metricName="r2")
    test_r2 = evaluator.evaluate(lr_predictions)
    print(f"测试集R2: {test_r2:.4f}")
    
    # 打印部分预测结果
    print("=== 部分预测样本 ===")
    lr_predictions.select("prediction", "col4", "features").show(5, truncate=False)

# 启动流式处理，用foreachBatch处理每个数据批次
streaming_query = stream_df \
    .writeStream \
    .foreachBatch(process_batch) \
    .start()

streaming_query.awaitTermination()

关键说明 #

1. foreachBatch机制：允许为每个到达的数据流批次执行自定义批处理逻辑，实现流式场景下的模型训练、评估操作。
2. 模型增量更新：通过全局变量global_model保存模型状态，首次批次初始化训练，后续批次用新数据迭代更新模型。
3. 空批次防护：加入空批次判断，避免无数据流入时触发无效计算。
4. 日志优化：设置日志级别为WARN，减少冗余日志干扰，聚焦模型输出内容。

内容的提问来源于stack exchange，提问作者Yannis