你的hdfs文件结构是这样的。。。

$tree path
path
├── country=abc
│   └── state=4
└── country=xyz
    ├── state=1
    ├── state=2
    ├── state=3
    ├── state=4
    ├── state=5
    └── state=6

您需要使用它来获取字符串形式的完整路径。。

val lb = new ListBuffer[String]
  def getAllFiles(path:String, sc: SparkContext):Unit = {
  val conf = sc.hadoopConfiguration
    val fs = FileSystem.get(conf)
    val files: RemoteIterator[LocatedFileStatus] = fs.listLocatedStatus(new Path(path))
    while(files.hasNext) {
      var filepath = files.next.getPath.toString
      //println(filepath)
      lb += (filepath)
      getAllFiles(filepath, sc)
    }
    println(lb)
  }

一旦你得到listbuffer所有文件的完整路径，包括子文件夹。。。您需要编写逻辑来填充Map。我要把它留给你。蒂伊。。
注意：listbuffer有一个组，您可以使用该组返回Map
在我的情况下，我做了这样的实验。。。

println( lb.groupBy(_.toString.replaceAll("file:/Users/xxxxxx/path/country=", "")substring(0, 3) ))

我得到的结果是

Map(abc -> ListBuffer(file:/Users/xxxxxx/path/country=abc, file:/Users/xxxxxx/path/country=abc/state=4), xyz -> ListBuffer(file:/Users/xxxxxx/path/country=xyz, file:/Users/xxxxxx/path/country=xyz/state=1, file:/Users/xxxxxx/path/country=xyz/state=6, file:/Users/xxxxxx/path/country=xyz/state=3, file:/Users/xxxxxx/path/country=xyz/state=4, file:/Users/xxxxxx/path/country=xyz/state=5, file:/Users/xxxxxx/path/country=xyz/state=2))

也许你可以用这个想法来进一步完善你想要的结果。