Spark批量读取Kafka遇null payload报错 如何过滤或记录空消息

问题根因 #

这个空指针异常的触发逻辑非常明确：

1. 自定义反序列化器未对入参data: Array[Byte]做非空判断，当Kafka消息payload为null时，null值直接传入Activity.validate方法，会直接抛出空指针，根本走不到定义的成功/失败分支返回None。
2. 旧版Kafka客户端在遇到null payload时，存在直接将ConsumerRecord的value字段赋值为null、不触发自定义反序列化逻辑的情况，此时直接调用flatMap(_.value())，Scala会尝试对null值做Option到Iterable的隐式转换，直接触发异常栈中option2Iterable相关的空指针。

Kafka中出现null payload通常有两种场景：一是开启压缩的主题生成的墓碑消息（tombstone），用于标记对应key的消息需要被清理；二是生产者端逻辑异常，误发送了value为null的无效消息。

修复方案 #

不需要修改Spark或Kafka的全局配置，两层代码防护即可同时实现null消息自动过滤、空消息元数据日志记录：

1. 修复自定义反序列化器，优先拦截空字节数组

在反序列化逻辑最开头增加null判断，确保空payload直接返回None，不会触发后续校验逻辑的空指针：

override def deserialize(topic: String, data: Array[Byte]): Option[Activity] = {
  // 优先处理null payload场景
  if (data == null) {
    return None
  }
  Activity.validate(data) match {
    case Success(value) => Some(value)
    case Failure(exp) =>
      // 可按需增加反序列化失败的错误日志
      None
  }
}

2. 调整RDD处理逻辑，增加兜底判空+日志打印

不要直接调用_.value()做flatMap，在算子外层初始化日志对象，处理逻辑中增加兜底判空，同时打印空消息的元数据方便后续排查问题：

import org.slf4j.LoggerFactory
// 日志对象必须初始化在算子外部，避免重复创建、序列化异常
val logger = LoggerFactory.getLogger(this.getClass)

val deviceRdd = KafkaUtils
  .createRDD[String, Option[Activity]](spark.sparkContext, kafkaParams, offsetRanges, PreferConsistent)
  .flatMap { record =>
    val recordValue = record.value()
    if (recordValue == null) {
      // 打印空消息的完整元数据，可直接用于回溯生产来源
      logger.warn(
        s"Found null payload message from Kafka, topic: ${record.topic()}, partition: ${record.partition()}, offset: ${record.offset()}, timestamp: ${record.timestamp()}"
      )
      None
    } else {
      recordValue
    }
  }

补充说明 #

• 两层校验可以100%覆盖null消息场景：反序列化层拦截空字节数组，处理层兜底拦截任何原因导致的value为null的记录，flatMap会自动过滤所有返回None的无效记录。
• 日志中打印的分区、偏移量等信息可以直接定位到具体消息，方便区分是正常的墓碑消息还是生产者bug导致的无效消息。

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