Spark 2.4.7从Kafka读取Avro：跳过格式错误记录及解决序列化问题

核心问题分析 #

你遇到的序列化错误，根源是外部定义的Column对象（decodedColumn）被引用到foreachBatch闭包中，而org.apache.spark.sql.Column未实现Serializable接口，导致Spark无法将闭包序列化分发到Executor执行。

同时你原有的过滤逻辑row.schema.equals(configuredStructTypeSchema)完全无效——Row的Schema由DataFrame定义，不会随单条记录的内容变化，根本无法筛选出格式正确的Avro数据。

正确解决方案 #

方案一：流层面直接处理（推荐，性能更优） #

将Avro解码和坏数据过滤逻辑直接在流DataFrame上完成，避免在foreachBatch闭包中引用外部不可序列化对象：

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

// 读取Kafka流
val kafkaStream = spark.readStream.format("kafka")
  .options(config.kafkaOptions ++ Map("subscribe" -> inputTopic))
  .load()

// 解码Avro数据，指定PERMISSIVE模式（解析失败返回null）
val decodedStream = kafkaStream.select(
  from_avro(
    $"value", 
    schema, 
    Map("mode" -> "PERMISSIVE") // 关键：让解析失败的记录返回null
  ) as "value"
)

// 过滤掉解析失败的行（value为null即格式错误）
val validStream = decodedStream
  .filter($"value".isNotNull)
  .select("value.*") // 展开结构化数据

// 输出处理后的流
val query = validStream.writeStream
  .option("checkpointLocation", checkpointLocation)
  .trigger(streaming.Trigger.ProcessingTime(outputTriggerTime))
  .foreachBatch { (df, batchId) =>
    writer.write(df)
  }.start

方案二：在foreachBatch内部处理（适用于必须在批处理阶段操作的场景） #

如果业务逻辑要求必须在foreachBatch内部完成解码，需在闭包内部创建from_avro的Column对象，避免引用外部不可序列化实例：

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

val query = spark.readStream.format("kafka")
  .options(config.kafkaOptions ++ Map("subscribe" -> inputTopic))
  .load()
  .writeStream
  .option("checkpointLocation", checkpointLocation)
  .trigger(streaming.Trigger.ProcessingTime(outputTriggerTime))
  .foreachBatch { (df, batchId) =>
    // 闭包内部创建解码Column，避免序列化问题
    val decodedColumn = from_avro(
      df("value"), 
      schema, 
      Map("mode" -> "PERMISSIVE")
    ) as "value"
    
    // 过滤有效数据并输出
    val validData = df.select(decodedColumn)
      .filter($"value".isNotNull)
      .select("value.*")
    writer.write(validData)
  }.start

关键说明 #

1. PERMISSIVE模式的作用：
   Spark 2.4的from_avro函数支持通过mode参数指定解析策略，PERMISSIVE模式会将格式错误的Avro记录解析为null，而非直接抛出异常，这是实现坏数据过滤的基础。
2. 避免Row级自定义函数：
   不要使用row => ...这种基于Row的过滤逻辑，这类操作无法被Spark优化，且极易引发序列化问题，优先使用Spark内置的列操作（如$"value".isNotNull）。
3. 序列化问题本质：
   foreachBatch闭包中的所有外部引用对象必须实现Serializable，Column对象不满足这一要求，因此必须将其创建逻辑移到闭包内部，或提前在流层面完成操作。

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