Spark单分区写入CSV无OOM报错的底层原理及实现方案问询

为什么数据量超过单Executor内存时任务没有崩溃 #

这一现象本质是由Spark写入流程的流式处理特性和内存管理机制决定的，核心原因如下：

• 写入任务不会全量加载全分区数据到内存：df.coalesce(1)生成的是窄依赖的单写入任务，该任务不需要先把2.7GB全量数据合并到内存后再写入，而是会依次迭代读取上游21个分区的小批量数据，每读取一批就直接编码为CSV格式写入磁盘，同一时间内存中仅保留当前处理的少量批量数据，完全不需要占用2.7GB的堆内存空间。
• Spark自带内存溢写机制：即使出现临时数据量超出Executor可用内存的情况，Spark会自动将超出部分的内存数据溢写到本地磁盘，待需要时再从磁盘读取，不会直接触发OOM错误。
• 源文件大小不等于内存占用大小：你看到的2.7GB是CSV文本的磁盘存储大小，Spark加载数据后会使用内部压缩的UnsafeRow格式存储，数值类型、日期类型等字段的内存占用会远低于文本格式的存储占用，实际内存开销远小于2.7GB。

超Executor内存场景下单分区写入CSV的实现方法 #

只要遵循以下规则，即可保证远大于Executor内存的数据集也能正常完成单分区写入：

• 合并分区时优先使用coalesce(1)，不要用repartition(1)：repartition(1)会触发全量shuffle，中间需要额外的shuffle内存开销，而coalesce(1)是窄依赖，直接流式拉取上游分区数据写入，内存开销极低。
• 不要在合并分区前对全量DataFrame执行cache()/persist()、全量聚合、全量去重等需要加载全量数据到内存的操作。如果业务逻辑必须做上述操作，建议先将处理后的结果写入临时磁盘路径，再重新读取临时文件执行coalesce(1).write操作。
• 可适当调低Spark SQL的批量处理参数，比如spark.sql.execution.arrow.maxRecordsPerBatch、spark.sql.files.maxPartitionBytes，降低单批次处理的数据量，进一步降低内存峰值。

内容的提问来源于stack exchange，提问作者sanchit08