目录中的Spark Scala列表文件夹

Question

我想使用Scala/Spark列出hdfs目录中的所有文件夹.在Hadoop中,我可以使用以下命令执行此操作:hadoop fs -ls hdfs://sandbox.hortonworks.com/demo/

我尝试过:

val conf = new Configuration()
val fs = FileSystem.get(new URI("hdfs://sandbox.hortonworks.com/"), conf)

val path = new Path("hdfs://sandbox.hortonworks.com/demo/")

val files = fs.listFiles(path, false)

但它似乎并没有在Hadoop目录中查找,因为我找不到我的文件夹/文件.

我也尝试过:

FileSystem.get(sc.hadoopConfiguration).listFiles(new Path("hdfs://sandbox.hortonworks.com/demo/"), true)

但这也无济于事.

你还有其他想法吗？

PS:我也检查过这个帖子:Spark迭代HDFS目录,但它对我不起作用,因为它似乎不在hdfs目录上搜索,而只在具有模式文件的本地文件系统上搜索//.

Answer 1

我们使用的是hadoop 1.4,它没有listFiles方法,因此我们使用listStatus来获取目录.它没有递归选项,但很容易管理递归查找.

val fs = FileSystem.get(new Configuration())
val status = fs.listStatus(new Path(YOUR_HDFS_PATH))
status.foreach(x=> println(x.getPath))

Answer 2

在 Spark 2.0+ 中，

import org.apache.hadoop.fs.{FileSystem, Path}
val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)
fs.listStatus(new Path(s"${hdfs-path}")).filter(_.isDir).map(_.getPath).foreach(println)

希望这是有帮助的。

Answer 3

 val listStatus = org.apache.hadoop.fs.FileSystem.get(new URI(url), sc.hadoopConfiguration)
.globStatus(new org.apache.hadoop.fs.Path(url))

  for (urlStatus <- listStatus) {
    println("urlStatus get Path:" + urlStatus.getPath())
}

Answer 4

我正在为S3寻找相同的东西，但不是HDFS。

我解决了使用我的 S3 路径创建文件系统的问题，如下所示：

  def getSubFolders(path: String)(implicit sparkContext: SparkContext): Seq[String] = {
    val hadoopConf = sparkContext.hadoopConfiguration
    val uri = new URI(path)

    FileSystem.get(uri, hadoopConf).listStatus(new Path(path)).map {
      _.getPath.toString
    }
  }

我知道这个问题与 HDFS 相关，但也许像我这样的其他人会来这里寻找 S3 解决方案。由于没有在 FileSystem 中指定 URI，它将查找 HDFS 的 URI。

java.lang.IllegalArgumentException: Wrong FS: s3://<bucket>/dummy_path
expected: hdfs://<ip-machine>.eu-west-1.compute.internal:8020

Answer 5

在 Ajay Ahujas 中，答案isDir已被弃用..

使用isDirectory...请参阅下面的完整示例和输出。

package examples

    import org.apache.log4j.Level
    import org.apache.spark.sql.SparkSession

    object ListHDFSDirectories  extends  App{
      val logger = org.apache.log4j.Logger.getLogger("org")
      logger.setLevel(Level.WARN)
      val spark = SparkSession.builder()
        .appName(this.getClass.getName)
        .config("spark.master", "local[*]").getOrCreate()

      val hdfspath = "." // your path here
      import org.apache.hadoop.fs.{FileSystem, Path}
      val fs = org.apache.hadoop.fs.FileSystem.get(spark.sparkContext.hadoopConfiguration)
      fs.listStatus(new Path(s"${hdfspath}")).filter(_.isDirectory).map(_.getPath).foreach(println)
    }

结果 ：

file:/Users/user/codebase/myproject/target
file:/Users/user/codebase/myproject/Rel
file:/Users/user/codebase/myproject/spark-warehouse
file:/Users/user/codebase/myproject/metastore_db
file:/Users/user/codebase/myproject/.idea
file:/Users/user/codebase/myproject/src