Scala Spark中如何为_fire和_water路径分别创建单个RDD

核心错误分析 #

导致RDD输出为空及代码异常的常见问题：

• 嵌套Seq结构：创建RDD时错误地将路径集合嵌套进另一层Seq，导致RDD类型变为RDD[Seq[String]]而非RDD[String]，输出无法正常解析
• displayFiles方法引用未定义变量：方法内使用了未声明的SparkContext、路径变量等
• 目录遍历逻辑漏洞：未正确收集到目标子目录路径，导致集合为空，最终RDD无数据

修正后的完整代码 #

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import java.io.File
import java.util.ArrayList

object DirectoryRDDHandler {
  def main(args: Array[String]): Unit = {
    // 初始化Spark配置与上下文
    val conf = new SparkConf().setAppName("DirRDDCreator").setMaster("local[*]")
    val sc = new SparkContext(conf)
    
    val rootDir = "/your/root/directory/path"
    val fireDirs = new ArrayList[String]()
    val waterDirs = new ArrayList[String]()
    
    // 遍历目录收集目标路径
    traverseDir(new File(rootDir), fireDirs, waterDirs)
    
    // 先验证收集到的路径（排查集合为空问题）
    println("已收集的_fire目录：")
    fireDirs.forEach(println)
    println("\n已收集的_water目录：")
    waterDirs.forEach(println)
    
    // 将Java ArrayList转换为Scala Seq，避免嵌套结构
    val firePathSeq = scala.collection.JavaConverters.asScalaBuffer(fireDirs).toSeq
    val waterPathSeq = scala.collection.JavaConverters.asScalaBuffer(waterDirs).toSeq
    
    // 分别创建对应类型的RDD
    val fireRDD = sc.parallelize(firePathSeq)
    val waterRDD = sc.parallelize(waterPathSeq)
    
    // 验证RDD输出
    println("\nFire RDD内容：")
    fireRDD.foreach(println)
    println("\nWater RDD内容：")
    waterRDD.foreach(println)
    
    // 调用修复后的displayFiles方法
    displayDirFiles(fireRDD, "Fire")
    displayDirFiles(waterRDD, "Water")
    
    sc.stop()
  }
  
  // 递归遍历目录，收集目标子路径
  def traverseDir(dir: File, fireDirs: ArrayList[String], waterDirs: ArrayList[String]): Unit = {
    if (dir.isDirectory) {
      val dirName = dir.getName
      if (dirName.endsWith("_fire")) {
        fireDirs.add(dir.getAbsolutePath)
      } else if (dirName.endsWith("_water")) {
        waterDirs.add(dir.getAbsolutePath)
      }
      // 递归处理子目录，用Option避免空指针
      Option(dir.listFiles()).getOrElse(Array.empty[File]).foreach { subFile =>
        traverseDir(subFile, fireDirs, waterDirs)
      }
    }
  }
  
  // 修复后的目录文件展示方法：明确传入RDD和类型标识，避免未定义变量
  def displayDirFiles(rdd: org.apache.spark.rdd.RDD[String], dirType: String): Unit = {
    println(s"\n=== ${dirType} 目录下的文件 ===")
    rdd.foreach { dirPath =>
      println(s"目录路径：${dirPath}")
      // 列出目录内的所有文件，用Option避免空指针
      val files = new File(dirPath).listFiles()
      Option(files).getOrElse(Array.empty[File]).foreach { file =>
        println(s"  - ${file.getName}")
      }
    }
  }
}

关键修正点说明 #

• 解决嵌套Seq问题：通过JavaConverters.asScalaBuffer将Java ArrayList转为Scala Buffer再转Seq，直接传入sc.parallelize，确保RDD的每个元素是单个字符串路径，而非嵌套的Seq集合
• 修复displayFiles方法：将目标RDD和类型标识作为参数传入，方法内所有变量均来自参数或内部声明，避免引用未定义的外部变量
• 增加路径验证步骤：在创建RDD前先打印收集到的目录路径，确认集合不为空，快速排查遍历逻辑是否存在问题
• 鲁棒性优化：遍历目录时用Option处理listFiles()可能返回的null值，避免空指针异常

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