PySpark千万级数据集执行Collect Action无返回且异常终止求助

• collect()的内存瓶颈：collect()会把分布式存储在所有节点的数据集全量拉取到Driver节点的内存中，转换成Python列表。1亿条记录哪怕每条仅占100字节，总大小也接近10GB，远超你设置的Driver内存（5GB），直接触发内存溢出（OOM），系统会强制终止进程，因此既无报错信息，也无法执行后续打印操作。
• Local模式的参数无效：你使用--master local[*]提交任务，这种模式下所有计算都在Driver进程内完成，--executor-memory参数完全不生效，实际可用内存仅为--driver-memory设置的5GB，进一步加剧了内存不足的问题。
• 代码笔误（非核心问题）：你贴出的代码存在两处错误：df.collet()应为df.collect()，print(len(l1))中的l1需改为list_of_items，不过这应该是输入时的笔误，毕竟你提到100万条记录时能正常打印。
• 小数据集正常的逻辑：100万条记录的内存占用远低于5GB，Driver节点可以轻松容纳，因此能正常执行。

如果需要后续处理数据集，不建议直接用collect()拉取全量数据，推荐以下替代方案：

1. 分布式处理（优先选择）：将后续操作全部放在Spark的DataFrame API中完成，比如使用filter()、map()、groupBy()等算子，让计算任务分散到各个节点执行，避免将全量数据拉取到Driver节点。
2. 分批拉取到本地：如果必须将数据拉到本地处理，可以分批执行：
   • 用limit()+offset()循环拉取：

     batch_size = 100000
     total_count = df.count()
     for i in range(0, total_count, batch_size):
         batch_df = df.limit(batch_size).offset(i)
         batch_list = batch_df.collect()
         # 处理当前批次的列表
         print(f"处理第{i//batch_size +1}批，共{len(batch_list)}条记录")
     

   • 用foreachPartition()在Executor节点上直接处理分区数据，无需拉取到Driver：

     def process_partition(partition):
         for record in partition:
             # 在这里处理单条记录
             pass
     df.foreachPartition(process_partition)
     

3. 临时调大Driver内存（不推荐）：如果一定要一次性拉取全量数据，可以将--driver-memory调整为40G左右（你的实例有64GB内存，需预留足够空间给系统进程），但这种方式极易导致节点崩溃，仅适合临时测试场景。

Spark的设计初衷是处理分布式大数据，避免将全量大数据拉取到单个Driver节点是最基本的使用原则，collect()仅适用于小数据集的调试或最终结果提取。此外，Local模式仅适合开发调试，生产环境建议使用集群模式（Standalone/Yarn/K8s），让Executor节点承担计算和存储压力。

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