scala - 使用Spark在hdfs中递归搜索所有文件名

Question

我一直在寻找一种方法来获取 Hadoop 文件系统 (hdfs) 中目录及其子目录中的所有文件名。

我发现我可以使用这些命令来获取它:

sc.hadoopConfiguration.set("mapreduce.input.fileinputformat.input.dir.recursive", "true")
sc.wholeTextFiles(path).map(_._1)

这是“wholeTextFiles”文档:

Read a directory of text files from HDFS, a local file system (available on all nodes), or any Hadoop-supported file system URI. Each file is read as a single record and returned in a key-value pair, where the key is the path of each file, the value is the content of each file.

Parameters:

path - Directory to the input data files, the path can be comma separated paths as the list of inputs.

minPartitions - A suggestion value of the minimal splitting number for input data.

Returns:

RDD representing tuples of file path and the corresponding file content

Note: Small files are preferred, large file is also allowable, but may cause bad performance., On some filesystems, .../path/* can be a more efficient way to read all files in a directory rather than .../path/ or .../path, Partitioning is determined by data locality. This may result in too few partitions by default.

如您所见，“wholeTextFiles”返回一对包含文件名及其内容的 RDD。所以我尝试映射它并只获取文件名，但我怀疑它仍在读取文件。

我怀疑是这样的原因:如果我尝试计数(例如)并且我得到相当于“内存不足”的 Spark (失去执行者并且无法完成任务)。

我宁愿使用 Spark 以最快的方式实现这一目标，但是，如果有其他具有合理性能的方式，我很乐意尝试一下。

编辑:要清除它——我想用 Spark 来做，我知道我可以用 HDFS 命令之类的东西来做——我想知道如何用 Spark 提供的现有工具来做这样的事情，也许我可以解释一下使“wholeTextFiles”不读取文本本身(有点像转换只在一个 Action 之后发生，而一些“命令”从未真正发生过)。

非常感谢!

Answer 1

这是列出所有文件直到最后一个子目录深度的方法....并且不使用 wholetextfiles并且是递归调用直到子目录的深度...

val lb = new scala.collection.mutable[String] // variable to hold final list of files
def getAllFiles(path:String, sc: SparkContext):scala.collection.mutable.ListBuffer[String] = {
 val conf = sc.hadoopConfiguration
 val fs = FileSystem.get(conf)
 val files: RemoteIterator[LocatedFileStatus] = fs.listLocatedStatus(new Path(path))
 while(files.hasNext) {// if subdirectories exist then has next is true
  var filepath = files.next.getPath.toString
  //println(filepath)
  lb += (filepath)
  getAllFiles(filepath, sc) // recursive call
 }
 println(lb)
 lb
}

就是这样。它经过了成功的测试。你可以按原样使用..