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

Question

我是Spark的新用户，我正在尝试处理坐在HDFS文件系统上的XML文件的大文件集。 1台计算机(实际上是VM)的“开发”群集上大约有15万个文件，总计约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()调用中加载文件。

我尝试了多种不同的方法，但都失败了:

使用call到SparkContext.textFile("/my/path")加载文件失败，因为SparkContext不可序列化

在Hadoop API中使用call到FileSystem.open(path)的情况下，在RDD之外实例化FileSystem失败，因为FileSystem无法序列化

使用从Hadoop API调用FileSystem.open(path)(在RDD内实例化FileSystem)失败，因为程序用完了文件句柄。

替代方法包括尝试使用 SparkContext.wholeTextFiles("/my/path/*")，因此我不必从map()调用中加载文件，但失败的原因是程序内存不足。据推测这是因为它渴望加载文件。

是否有人在自己的工作中尝试过类似的尝试？如果是，您使用了哪种方法？

Answer 1

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

实际上，spark可以读取整个hfs目录，引自spark documentation

All of Spark’s file-based input methods, including textFile, support running on directories, compressed files, and wildcards as well. For example, you can use textFile("/my/directory"), textFile("/my/directory/*.txt"), and textFile("/my/directory/*.gz").