apache-spark - 并行加载多个文件的最有效方法是什么？

Question

[Disclaimer: While this question is somewhat specific, I think it circles a very generic issue with Hadoop/Spark.]

我需要在 Spark 中处理一个大型数据集(~14TB)。不做聚合，主要是过滤。给定约 30k 个文件(250 个部分文件，每月 10 年，每个部分约 200MB)，我想将它们加载到 RDD/DataFrame 中并根据一些任意过滤器过滤掉项目。

为了使文件列表高效(我在 google dataproc/cloud storage，所以执行通配符 glob 的驱动程序非常串行且非常慢)，我预先计算了文件名的 RDD，然后将它们加载到 RDD 中(我正在使用 avro，但文件类型不应该相关)，例如

#returns an array of files to load
files = sc.textFile('/list/of/files/').collect()  

#load the files into a dataframe
documents = sqlContext.read.format('com.databricks.spark.avro').load(files)

当我这样做时，即使在 50 个工作人员的集群上，似乎也只有一个执行程序在做读取文件的工作。我已经尝试过广播文件列表并阅读了十几种不同的方法，但我似乎无法解决这个问题。

那么，有没有一种有效的方法可以从多个文件创建一个非常大的数据框？在创建此 RDD 时，如何最好地利用所有潜在的计算能力？

这种方法在较小的集合上效果很好，但是在这种规模下，我看到了大量的症状，比如没有反馈的长时间运行的进程。是否有一些知识宝库——除了@zero323 :-)——以这种规模优化 Spark ？

Answer 1

列出 30k 文件对于 GCS 来说应该不是问题 - 即使一次列出多达 500 个文件的单个 GCS 列表请求每次需要 1 秒，所有 30k 文件将在一分钟左右列出。可能存在一些带有一些 glob 模式的极端情况，使其速度变慢，但最近在 GCS 连接器中进行了优化 globbing implementation这可能会有所帮助。

这就是为什么只依赖默认的 Spark API 和 globbing 就足够了:

val df = sqlContext.read.avro("gs://<BUCKET>/path/to/files/")