以分布方式在Spark中读取CSV文件

Question

我正在开发一个Spark处理框架，它读取大型CSV文件，将它们加载到RDD中，执行一些转换并在最后保存一些统计信息。

有问题的CSV文件平均大约为50GB。我正在使用Spark 2.0。

我的问题是：

当我使用sparkContext.textFile（）函数加载文件，是否需要将文件保存在驱动程序的内存第一，然后将其分配给工人（因此在驱动程序上需要相当大的内存）？或者文件被每个工作人员“并行”读取，这些文件都不需要存储整个文件，驱动程序只能作为“管理员”使用？

在此先感谢

Answer 1

当你定义的阅读，文件将被划分根据您的并行方案分区，并指示将被发送到工人。然后，文件由文件系统中的工作人员直接读取（因此需要为所有节点（如HDFS）提供分布式文件系统）。

作为一个方面说明，使用spark.read.csv而不是RDD将其读取到数据框会好得多。这将花费更少的内存，并允许火花优化您的查询。

UPDATE

在评论，有人问，如果文件系统没有分布式文件将位于只有一台机器上会发生什么。 答案是，如果你有一台以上的机器，它很可能会失败。

当您执行sparkContext.textFile时，实际上没有任何内容被读取，它只是告诉spark要读取什么。然后你对它进行一些改造，因为你正在定义一个计划，所以还没有任何东西被读取。一旦你执行了一个动作（例如收集），那么实际的处理就开始了。 Spark将工作划分为任务并将其发送给执行人员。执行者（可能在主节点或工作者节点上）然后尝试读取文件的一部分。问题是任何不在主节点上的执行程序都会查找该文件，并且无法找到导致任务失败的文件。 Spark会重试几次（我相信默认值是4），然后完全失败。

当然，如果你只有一个节点，那么所有的执行者都会看到这个文件，一切都会好的。在理论上，也可能是任务在工人身上失败，然后重新运行主人并在那里成功，但无论如何，除非他们看到文件的副本，否则工人不会做任何工作。

您可以通过将文件复制到所有节点中完全相同的路径或通过使用任何类型的分布式文件系统（甚至NFS共享都可以）来解决此问题。

当然，您可以始终在单个节点上工作，但您不会利用spark的可扩展性。