Spark 将数据拉入 RDD 或数据框或数据集答案

【问题标题】：Spark pulling data into RDD or dataframe or datasetSpark 将数据拉入 RDD 或数据框或数据集
【发布时间】：2016-12-27 06:35:45
【问题描述】：

我试图简单地说，当 spark 通过驱动程序提取数据时，以及当 spark 不需要通过驱动程序提取数据时。

我有 3 个问题 -

假设您有一个存储在 HDFS 中的 20 TB 平面文件文件，您可以使用相应库的开箱即用功能之一（sc.textfile(path) 或 @ 987654322@等）。如果驱动程序仅使用 32 GB 内存运行，是否会导致驱动程序出现 OOM？或者至少有司机吉姆的交换？或者 spark 和 hadoop 是否足够聪明，可以将来自 HDFS 的数据分发到 spark 执行器中，从而在不通过驱动程序的情况下生成数据帧/RDD？
除了来自外部 RDBMS 之外，与 1 完全相同的问题？
除了来自特定节点文件系统（仅 Unix 文件系统，20 TB 文件但不是 HDFS）之外，与 1 完全相同的问题？

【问题讨论】：

我已经看到前两个..大数据，大于驱动程序内存没有给出任何错误。所以是的，前两个。如果说第三个不起作用，您可以随时将其推送到 HDFS，然后这将是第一个问题。事实上，尽管数字从 1 到 3，但您有 2 个问题。

【解决方案1】：

关于1

Spark 使用分布式数据结构，如 RDD 和 Dataset（以及 2.0 之前的 Dataframe）。为了回答您的问题，您应该了解以下有关此数据结构的事实：

所有转换操作，如（映射、过滤器等）都是惰性的。这意味着除非您需要操作的具体结果（如减少、折叠或保存结果到某个文件）。
在 HDFS 上处理文件时，Spark 运行与文件分区。分区是最小的逻辑批数据可以处理的。通常一个分区等于一个HDFS 块和分区的总数永远不能少文件中的块数。常见的（也是默认的）HDFS 块大小为 128Mb
RDD 中的所有实际计算（包括从 HDFS 读取）和数据集在执行器内部执行，从不在驱动程序上执行。司机创建 DAG 和逻辑执行计划并将任务分配给执行人进行进一步处理。
每个执行器运行之前的针对特定数据分区分配的任务。因此，通常如果您只为执行程序分配一个核心，它会同时处理不超过 128Mb（默认 HDFS 块大小）的数据。

所以基本上当您调用sc.textFile 时，不会发生实际读取。所有提到的事实都解释了为什么在处理 20 Tb 的数据时也不会发生 OOM。

有一些特殊情况，例如 join 操作。但即使在这种情况下，所有 executor 都会将它们的中间结果刷新到本地磁盘以供进一步处理。

如果是 JDBC，您可以决定您的表有多少个分区。并在表中选择适当的分区键，将数据正确拆分为分区。由您决定将多少数据同时加载到内存中。

本地文件的块大小由fs.local.block.size 属性控制（我猜默认为32Mb）。因此它与 1（HDFS 文件）基本相同，只是您将从一台机器和一个物理磁盘驱动器读取所有数据（这在 20TB 文件的情况下效率极低）。

【讨论】：

我在我拥有的集群上进行了测试，我确实从 HDFS 中提取了一个更大的数据文件，并从 postgre 实例中提取了更大的表到我的 spark 集群中，没有任何问题。不是 TB，而是超过 40 GB。我也将一个本地文件拉入 Spark，它也可以工作，但我不能只在一台机器上拥有它。我需要将 hte 40 GB 文件放在所有有执行程序的机器上。
我的解释总结是Spark永远不会将整个数据集加载到内存中。只有一小块。要在集群中分发 40Gb 文件，您可以使用 HDFS。在这种情况下，Spark 会根据数据局部性将任务分配给执行者。这意味着执行器将仅处理位于执行器运行的同一台机器上的那些块（在最佳情况下）。
我已经测试了第 1 部分和第 2 部分，是的，它从 rdbms 和 hdfs 中提取了它，并且文件和表比执行器放在一起的所有内存大得多。只有当整个数据集返回到应用程序内的 java 数组时，驱动程序才会发挥作用。感谢您的回答。当然我可以把数据放到hdfs中分发。我必须制作文件的 3 个本地副本才能从本地驱动器将其读入 spark，这不是我开始测试时预期的问题。
关于驱动程序 - 是的，您完全正确。如果将所有数据收集到驱动程序，它可能真的会耗尽内存。这对于任何具有未绑定结果的操作（如take 和countByKey）也是如此。很高兴我的回答对您有所帮助。