为何try/catch未捕获NPE致Spark流写入Cassandra程序终止？

问题根源：Spark的分布式执行逻辑 #

你写的try/catch块是在Driver端执行的，但saveToCassandra的实际数据写入逻辑是跑在集群的Executor节点上的。Spark Streaming的DStream操作会把处理任务分发到各个Executor上并行执行，Executor抛出的异常不会被Driver端的try/catch捕获——这些异常会触发Spark的任务失败机制，直接导致整个作业终止，根本不会进入你写的catch代码块。

从你提供的异常栈也能印证这一点：异常最终出现在org.apache.spark.executor.Executor$TaskRunner.run这一层，完全属于Executor的任务执行线程，和Driver端的代码栈是分离的。

解决办法：从源头清理数据，避免null值进入Cassandra #

Cassandra的map<text, text>类型确实不允许值为null，所以最稳妥的方式是在写入前就对流数据中的map做清洗：

方案1：替换或移除map中的null值 #

假设你的DStream元素是一个包含map字段的Case Class（比如case class MyData(id: String, infoMap: Map[String, String])），可以在saveToCassandra前添加map操作清理数据：

// 方式A：把null值替换为空字符串
val cleanedStream = cassandraStream.map { data =>
  val cleanedMap = data.infoMap.mapValues(value => Option(value).getOrElse(""))
  data.copy(infoMap = cleanedMap)
}

// 方式B：直接移除值为null的键值对
val cleanedStream = cassandraStream.map { data =>
  val cleanedMap = data.infoMap.filter { case (_, value) => value != null }
  data.copy(infoMap = cleanedMap)
}

// 清洗完成后再写入Cassandra
cleanedStream.saveToCassandra("keyspace", "table")

方案2：在Executor端做单条数据的异常捕获（可选） #

如果需要保留异常记录逻辑，可以通过foreachRDD在每个分区的Executor端捕获异常，但这种方式不如提前清洗高效，因为异常会导致部分任务重试或失败：

cassandraStream.foreachRDD { rdd =>
  rdd.foreachPartition { partition =>
    val connector = CassandraConnector(sc.getConf)
    connector.withSessionDo { session =>
      val preparedStmt = session.prepare("INSERT INTO keyspace.table(id, info_map) VALUES (?, ?)")
      partition.foreach { data =>
        try {
          // 先清理map再执行写入
          val cleanedMap = data.infoMap.mapValues(Option(_).getOrElse(""))
          val boundStmt = preparedStmt.bind(data.id, cleanedMap)
          session.execute(boundStmt)
        } catch {
          case e: Exception => 
            // 这里的catch是在Executor端执行，可以记录单条数据的写入错误
            println(s"Failed to write data ${data.id}: ${e.getMessage}")
        }
      }
    }
  }
}

总结 #

别指望Driver端的try/catch能处理Executor端的任务异常——Spark的分布式模型决定了这种方式完全无效。最可靠的做法是在数据写入Cassandra之前，彻底清理掉不符合要求的null值，从源头避免异常发生。

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