Spark执行foreachPartition报Task not serializable及后续异常求助

你猜测的各类writer不可序列化是错误判断：这些writer是在Executor端分区执行逻辑内部初始化的，不需要经过序列化传输，不会触发序列化校验。
真正的原因是传入foreachPartition的闭包捕获了外部类的实例引用：
你代码中用到的localLuceneIndexDirPath、PROTOS_CACHE_FILE、PROTOS_MD_FILE如果是外部类的成员变量，或者getIndexWriter是外部类的成员方法，Scala会自动将整个外部类的引用打包到闭包里，序列化后发送到Executor。只要外部类没有实现Serializable，就会触发Task not serializable异常。

异常原因 #

你强行给相关类添加Serializable后出现no valid constructor错误，通常是因为你的类是内部类、匿名类，或者没有无参构造函数，Java反序列化时无法实例化对应类；连续多次任务失败后触发了Spark的黑名单机制，会将失败次数超标的Executor、节点加入黑名单，最终导致没有可用资源调度任务。

解决步骤 #

• 优先解决序列化根本问题，不要强行给业务类加Serializable #

  • 将闭包内用到的所有外部参数、常量提前提取为Driver端的局部变量，或者通过广播变量传递，避免直接引用外部类成员，示例调整如下：

    // Driver端提前提取为局部变量，不要引用外部类成员
    val localDir = localLuceneIndexDirPath
    val cacheFileName = PROTOS_CACHE_FILE
    val mdFileName = PROTOS_MD_FILE
    documents.repartition(1).foreachPartition( allDocuments => {
      val luceneIndexWriter: IndexWriter = getIndexWriter(localDir)
      val protosCache = Files.newOutputStream(Paths.get(s"${localDir}/${cacheFileName}"))
      val protosMdFile = Files.newOutputStream(Paths.get(s"${localDir}/${mdFileName}"))
      val DOCID: AtomicInteger = new AtomicInteger(1)
      val umcIdsCache = new mutable.ListBuffer[String] 
    
      allDocuments.foreach ( row => {
      .....
      })
      luceneIndexWriter.commit()
      luceneIndexWriter.close()
    })
    

  • 将getIndexWriter这类工具方法抽到独立的静态对象（Scala的object、Java的static类）中，避免方法依赖类实例，减少闭包捕获的内容。
  • 对类中不需要序列化的成员添加@transient注解，告知Spark序列化时忽略该字段。

• 修复no valid constructor错误 #

  • 不要使用非静态内部类编写分区处理逻辑，将业务逻辑抽到独立的顶层工具类中，避免内部类持有外部类引用。
  • 如果你使用的是Scala的object单例对象，不需要手动添加Serializable声明，Scala原生object默认支持序列化，仅需给内部非序列化成员加@transient即可。

• 解决节点/Executor黑名单问题 #

  • 该问题是任务连续失败触发的衍生问题，修复上述序列化错误后，任务不再连续失败，黑名单问题会自动消失。
  • 测试环境可以临时关闭黑名单规避：将Spark参数spark.blacklist.enabled设为false，或者调大spark.task.maxFailures参数的阈值，允许更多任务重试次数（生产环境不建议关闭黑名单）。
  • 排除基础资源问题：检查Yarn节点的内存、磁盘、网络状态，确认没有资源不足导致的任务意外失败。

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