使用 Spark API 如何处理大型目录树？

Question

我是新的 Spark 用户，我正在尝试处理位于 HDFS 文件系统上的大量 XML 文件集。在 1 台机器（实际上是 VM）的“开发”集群上大约有 150k 个文件，总计约 28GB。

这些文件在 HDFS 中被组织成一个目录结构，这样在一个父目录下就有大约一百个子目录。每个“子”目录包含几百到几千个 XML 文件之间的任何内容。

我的任务是解析每个 XML 文件，使用 XPath 表达式提取一些值，然后将结果保存到 HBase。我正在尝试使用 Apache Spark 来做到这一点，但我运气不佳。我的问题似乎是 Spark API 和 RDD 工作方式的结合。在这一点上，分享一些伪代码来表达我正在尝试做的事情可能是谨慎的：

RDD[String] filePaths = getAllFilePaths()
RDD[Map<String,String>] parsedFiles = filePaths.map((filePath) => {
    // Load the file denoted by filePath
    // Parse the file and apply XPath expressions
})
// After calling map() above, I should have an RDD[Map<String,String>] where
// the map is keyed by a "label" for an xpath expression, and the
// corresponding value is the result of the expression applied to the file 

所以，暂时忽略我写给 HBase 的部分，让我们专注于上述内容。我无法从 RDD map() 调用中加载文件。

我尝试了很多不同的方法，但都失败了：

1. 使用call 到SparkContext.textFile("/my/path") 加载文件失败，因为SparkContext 不可序列化
2. 使用 Hadoop API 中的 call 到 FileSystem.open(path)，其中 FileSystem 在 RDD 外部实例化失败，因为 FileSystem 不可序列化
3. 使用从 Hadoop API 调用 FileSystem.open(path)，其中 FileSystem 在内部被实例化，RDD 失败，因为程序用完了文件句柄。

替代方法包括尝试使用SparkContext.wholeTextFiles("/my/path/*")，因此我不必从 map() 调用中加载文件，因为程序内存不足而失败。这大概是因为它急切地加载文件。

有没有人在他们自己的工作中尝试过类似的事情，如果有，您使用了什么方法？

Answer 1

尝试使用通配符读取整个目录。 val errorCount = sc.textFile("hdfs://some-directory/*")

其实spark可以读取整个hfs目录，引用自sparkdocumentation

Spark 的所有基于文件的输入法，包括 textFile，都支持 也可以在目录、压缩文件和通配符上运行。为了 例如，您可以使用textFile("/my/directory")， textFile("/my/directory/*.txt") 和 textFile("/my/directory/*.gz")。