如果整个文件存在于单个节点上，火花将如何加载巨大的 csv 文件？

Question

如果我在单个 HDFS 节点上有一个 50GB 的巨大 CSV 文件，并且我正在尝试使用 spark.read 读取该文件，如下所示：

file_df = spark.read.format('csv').option('header', 'true').option('inferSchema', 'true').load('/hdfspath/customer.csv')

我正在使用以下 spark-submit 提交 spark 作业：

spark-submit --master yarn --deploy-mode cluster --num-executors 4 --executor-memory 3G --executor-cores 5 --driver-memory 1G load_csv.py

我知道在有动作之前，spark 不会将任何数据加载到内存中。但是当一个动作被触发并且要做的第一件事就是将文件读入内存以启动转换时会发生什么。 spark如何根据我提到的core和executors读取50GB的小文件？

例如：我提到了 4 个执行器和每个执行器的 3GB 内存。读取时，将为每个执行程序将主 customer.csv 文件转换为 3GB 的块并加载文件：

对于前 12GB：

Executor 1: 3GB
Executor 2: 3GB
Executor 3: 3GB
Executor 4: 3GB

以此类推，直到整个文件处理完毕？

或者它会根据 HDFS 块大小拆分文件并逐块读取，例如：128MB，它会尝试在每个 3GB 的执行程序中尽可能多地适应块？

如果文件完全存在于单个集群上（在我的例子中是这样），spark 将如何处理文件？

我理解它的解释有点宽泛和麻烦，但任何帮助将不胜感激。

Answer 1

IIUC，

这些是众所周知的调整 spark 处理大型数据集的一般做法(50 GB is not a huge dataset either)

1. 它是否会根据 HDFS 块大小拆分文件并逐块读取，例如：128MB，并且它会尝试在每个 3GB 的执行器中尽可能多地容纳块？

回答：是的， Spark 将根据1 partition for 1 HDFS block(128 MB ideally) for splitable fileformats 创建分区，而不是在这种情况下根据执行程序内存。

2。内存和磁盘中的存储级别`

在缓存上（与persist(StorageLevel.MEMORY_ONLY) 相同，它会将所有分区存储在内存中 - 如果它不适合内存，您将收到 OOM。如果您调用 persist(StorageLevel.MEMORY_AND_DISK)，它将尽可能多地存储在内存中，其余的将放在磁盘上。如果数据不适合磁盘，操作系统通常会杀死您的工作人员。

请注意，Spark 有自己的小内存管理系统。您分配给 Spark 作业的一些内存用于保存正在处理的数据，如果您调用缓存或持久化，则一些内存用于存储。

from pyspark.storagelevel import StorageLevel

file_df = spark.read.format('csv').option('header', 'true')
.option('inferSchema', 'true').load('/hdfspath/customer.csv')
import org.apache.spark.storage.StorageLevel

file_df = file_df.persist(StorageLevel.MEMORY_AND_DISK)
//val df2 = df.persist(StorageLevel.DISK_ONLY)

Storage Level    Space used  CPU time  In memory  On-disk  Serialized   Recompute some partitions
----------------------------------------------------------------------------------------------------
MEMORY_ONLY          High        Low       Y          N        N         Y    
MEMORY_ONLY_SER      Low         High      Y          N        Y         Y
MEMORY_AND_DISK      High        Medium    Some       Some     Some      N
MEMORY_AND_DISK_SER  Low         High      Some       Some     Y         N
DISK_ONLY            Low         High      N          Y        Y         N

3.尝试以下选项进行内存选项设置。

spark-submit --master yarn --deploy-mode cluster --num-executors ex4 --executor-memory 3G --executor-cores 5 --driver-memory 3G load_csv.py

假设您有 10 个节点集群，具有以下配置，

**Cluster Config:**
10 Nodes
16 cores per Node
64GB RAM per Node

3.1 第一种方法：微型执行器[每个核心一个执行器]：

- `--num-executors` = `In this approach, we'll assign one executor per core`
                    = `total-cores-in-cluster`
                   = `num-cores-per-node * total-nodes-in-cluster` 
                   = 16 x 10 = 160
- `--executor-cores` = 1 (one executor per core)
- `--executor-memory` = `amount of memory per executor`
                     = `mem-per-node/num-executors-per-node`
                     = 64GB/16 = 4GB

如上所述，每个核心只有一个执行器，我们将无法利用在同一个 JVM 中运行多个任务的优势。此外，广播变量和累加器等共享/缓存变量将在节点的每个核心中复制 16 次。此外，我们没有为 Hadoop/Yarn 守护进程留下足够的内存开销，并且我们没有计入 ApplicationManager。不好！

3.2 第二种方法：胖执行器（每个节点一个执行器）：

- `--num-executors` = `In this approach, we'll assign one executor per node`
                    = `total-nodes-in-cluster`
                   = 10
- `--executor-cores` = `one executor per node means all the cores of the node are assigned to one executor`
                     = `total-cores-in-a-node`
                     = 16
- `--executor-memory` = `amount of memory per executor`
                     = `mem-per-node/num-executors-per-node`
                     = 64GB/1 = 64GB

每个 executor 全部 16 个核心，除了 ApplicationManager 和守护进程不计算在内，HDFS 吞吐量会受到影响，并且会导致过多的垃圾结果。另外，不好！

3.3 第三种方法：Fat (vs) Tiny 之间的平衡

根据上述建议，

**1. Cores**
Let’s assign 5 core per executors => `--executor-cores = 5 (for good HDFS throughput)`
Leave 1 core per node for Hadoop/Yarn daemons => `Num cores available per node = 16-1 = 15`
So, Total available of cores in cluster = 15 x 10 = 150

**2. Executors**
Number of available executors = `(total cores/num-cores-per-executor) = 150/5 = 30`
Leaving 1 executor for ApplicationManager => --num-executors = 29
Number of executors per node = 30/10 = 3
Memory per executor = 64GB/3 = 21GB
Counting off heap overhead = 7% of 21GB = 3GB. So, actual --executor-memory = 21 - 3 = 18GB

因此，推荐的配置是：29 executors、18GB memory each 和 `5 个核心，用于上述 10 节点集群

--num-executors、--executor-cores 和 --executor-memory 这三个参数在 Spark 性能中起着非常重要的作用，因为它们控制着你的 Spark 应用程序获得的 CPU 和内存量。这使得用户了解正确的配置方式非常重要。