Spark程序运行遭遇三类异常：文件已存在、不支持操作、序列化失败

问题背景 #

编写的Spark程序包含两个核心步骤：

1. 基于源RDD计算字符串集合tag_set，通过collectAsMap().keySet()收集到Driver端，再封装为Broadcast<Set<String>>广播变量；
2. 利用该广播变量对源RDD执行map转换，最终将结果保存至HDFS路径。

问题一：提前清理输出路径仍报FileAlreadyExistsException #

原因

1. 清理不彻底：仅删除路径本身，但路径下残留子文件/目录（未递归删除），HDFS仍判定路径存在；
2. 元数据同步延迟：HDFS是分布式系统，Driver端执行清理后，集群元数据未及时同步，Spark检查路径时仍识别为存在；
3. 路径不一致：清理的路径与Spark输出路径存在细微差异（如大小写、末尾斜杠、相对/绝对路径混淆）；
4. 并行执行冲突：清理操作与Spark写操作未完全串行，Spark检查路径时清理动作尚未完成。

解决方法

• 优先使用Spark内置覆盖模式，无需手动清理：在保存结果时添加mode(SaveMode.Overwrite)，示例：

  resultRDD.saveAsTextFile(outputPath, SaveMode.Overwrite)
  

• 若必须手动清理，确保递归删除并确认路径不存在：

  val fs = FileSystem.get(sc.hadoopConfiguration)
  val path = new Path(outputPath)
  if (fs.exists(path)) {
    fs.delete(path, true) // true表示递归删除所有子内容
    // 可选：等待HDFS元数据同步
    while (fs.exists(path)) {
      Thread.sleep(1000)
    }
  }
  

• 严格核对清理路径与输出路径，确保完全一致。

问题二：添加HDFS删除代码后报UnsupportedOperationException（指向广播变量contains方法） #

原因

collectAsMap().keySet()返回的是原Map的视图集合（AbstractMap$KeySet），并非独立的可序列化集合。该视图依赖原Map对象，广播时仅序列化视图引用，在Executor端反序列化后原Map已不存在，调用contains()时触发UnsupportedOperationException。

解决方法

将keySet转换为独立的可序列化集合（如HashSet）后再封装广播变量：

// 先转为独立HashSet，再广播
val tag_set = new HashSet[String](rdd.collectAsMap().keySet())
val global_tag_set = sc.broadcast(tag_set)

问题三：直接使用tag_set报NotSerializableException（提示AbstractMap$1无法序列化） #

原因

collectAsMap().keySet()返回的视图集合（AbstractMap$KeySet）是Java AbstractMap的内部类，未实现Serializable接口。当在RDD的map算子中直接引用该变量时，Spark需要将其序列化发送至Executor，因无法序列化而抛出异常。

解决方法

与问题二解决方案一致，将keySet转换为独立的可序列化集合：

// 转换为HashSet后再在map中使用
val tag_set = new HashSet[String](rdd.collectAsMap().keySet())
val resultRDD = sourceRDD.map(e => {
  if(tag_set.contains(e.getLabel)) {
    // 业务逻辑处理
  }
})

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