scala - Spark从具有相同名称的所有子文件夹中递归读取文件

Question

我有一个过程，每小时将大量数据推送到 Blob 存储，并在我的存储容器中创建以下文件夹结构，如下所示:/year=16/Month=03/Day=17/Hour=16/mydata.csv
/year=16/Month=03/Day=17/Hour=17/mydata.csv

等等

在我的 Spark 上下文中形成我想访问所有 mydata.csv 并处理它们。我发现我需要设置 sc.hadoopConfiguration.set("mapreduce.input.fileinputformat.input.dir.recursive","true") 以便我们可以使用递归搜索，如下所示:

val csvFile2 = sc.textFile("wasb://mycontainer@mystorage.blob.core.windows.net/*/*/*/mydata.csv")

但是当我执行以下命令查看我收到了多少文件时，它给了我一些非常大的数字，如下所示

csvFile2.count
res41:长 = 106715282
理想情况下它应该返回 24*16=384，我也在容器上验证过，它只有 384 个 mydata.csv 文件，但由于某些原因我看到它返回 106715282。

谁能帮我弄清楚哪里出错了？

问候基兰

Answer 1

SparkContext 有两个相似的方法:textFile 和 wholeTextFiles。

textFile 加载每个文件的每一 行 作为 RDD 中的一条记录。因此 count() 将返回所有文件的总行数(在大多数情况下，例如您的文件，这将是一个很大的数字)。

wholeTextFiles 加载每个整个文件 作为 RDD 中的一条记录。因此 count() 将返回文件总数(在您的情况下为 384)。