csv - 以分布式方式读取Spark中的CSV文件

Question

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

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

我的问题是:

当我使用 sparkContext.textFile() 函数加载文件时，是否需要先将文件存储在驱动程序的内存中，然后将其分发给工作人员(因此需要驱动程序上相当大的内存)？或者每个工作人员都“并行”读取文件，在某种程度上，他们都不需要存储整个文件，而驱动程序仅充当“管理器”？

提前致谢

Answer 1

当您定义读取时，文件将根据您的并行方案划分为分区，并将指令发送给工作人员。然后工作人员直接从文件系统读取文件(因此需要一个分布式文件系统可用于所有节点，例如 HDFS)。

作为旁注，使用 spark.read.csv 而不是在 RDD 中将其读取到数据帧会好得多。这将占用更少的内存，并允许 spark 优化您的查询。

更新

在评论中，有人问如果文件系统没有分发并且文件只位于一台机器上会发生什么。
答案是，如果您有超过 1 台机器，它很可能会失败。

当您执行 sparkContext.textFile 时，实际上没有读取任何内容，它只是告诉 spark 您想要读取的内容。然后您对其进行一些转换，但仍然没有读取任何内容，因为您正在定义一个计划。一旦你执行了一个 Action (例如收集)，那么实际的处理就开始了。 Spark 会将作业划分为任务并将它们发送给执行者。执行器(可能在主节点或工作节点上)然后会尝试读取文件的一部分。问题在于，不在主节点上的任何执行程序都会查找该文件，但无法找到它，从而导致任务失败。 Spark 会重试几次(我相信默认值为 4)然后完全失败。

当然，如果您只有一个节点，那么所有执行程序都会看到该文件，一切都会好起来的。同样在理论上，任务可能会在 worker 上失败，然后在 master 上重新运行并在那里成功，但在任何情况下，worker 都不会做任何工作，除非他们看到文件的副本。

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

当然，您始终可以在单个节点上工作，但这样您就无法利用 Spark 的可扩展性。