使用sc.textFile递归获取子目录中的文件内容

Question

看起来SparkContext的textFile只期望给定目录位置中存在文件 - 它不会

• (a) 递归或者
• (b) 支持目录（尝试将目录读取为文件）

有什么建议如何构建一个递归 - 可能比手动创建递归文件列表/下降逻辑更简单？

这是用例：在

/data/tables/my_table

下面的所有目录级别中的所有文件都希望能够通过hdfs调用进行读取。

更新

sc.textFile()通过（子类）TextInputFormat调用Hadoop FileInputFormat。内部逻辑确实存在于执行递归目录读取的逻辑 - 即首先检测条目是否为目录，如果是，则进行下降：

<!-- language: java -->
     for (FileStatus globStat: matches) {
218          if (globStat.isDir()) {
219            for(FileStatus stat: fs.listStatus(globStat.getPath(),
220                inputFilter)) {
221              result.add(stat);
222            }          
223          } else {
224            result.add(globStat);
225          }
226        }

然而，当调用sc.textFile时，目录条目会出现错误：“不是文件”。这种行为很令人困惑-因为似乎已经有适当的支持来处理目录了。

- WestCoastProjects

---

一个可能的解决方案：https://dev59.com/fF4c5IYBdhLWcg3wc6Dr - Justin Pihony

@JustinPihony 谢谢，我没有使用S3 - 所以不清楚那个答案是否适用。 - WestCoastProjects

你能提供一个目录布局的例子吗？难道通配符语法不够用吗？例如：textFile(/path/*/*)。 - Nick Chammas

@NickChammas 不，通配符匹配不起作用：它不会下降，而且遇到的任何目录都会生成错误“不是文件”。 - WestCoastProjects

Answer 1

我正在查看旧版本的FileInputFormat..

BEFORE设置递归配置mapreduce.input.fileinputformat.input.dir.recursive

scala> sc.textFile("dev/*").count
     java.io.IOException: Not a file: file:/shared/sparkup/dev/audit-release/blank_maven_build

默认值为 null/未设置，被评估为“false”：

scala> sc.hadoopConfiguration.get("mapreduce.input.fileinputformat.input.dir.recursive")
res1: String = null

之后：

现在设置数值：

sc.hadoopConfiguration.set("mapreduce.input.fileinputformat.input.dir.recursive","true")

现在重新尝试递归操作：

scala>sc.textFile("dev/*/*").count

..
res5: Long = 3481

So it works.

更新: 根据@Ben的评论，添加了/以进行完全递归。

- WestCoastProjects

---

非常有帮助。值得补充的是，每个级别都需要一个星号。我的数据存储方式更像是 dev/<year>/<month>/<day>。要获取多个月的数据，我需要使用类似于 sc.textFile("dev/2015/*/*").count 的语法。 - jbencook

谢谢您的额外评论。我会更新我的帖子。 - WestCoastProjects

我正在使用Spark 1.6.1，不需要在每个级别上使用星号。 - Barbaros Alp

哇！有人在这里添加了一个无偿赏金。以前从未见过这样的情况。 - WestCoastProjects

我在 fileStream API 上进行了测试，但它没有起作用。我有什么遗漏吗？ - Gary Gauh

Answer 2

我发现这些参数必须按以下方式设置：

.set("spark.hive.mapred.supports.subdirectories","true")
.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive","true")

- Paul