PySpark写入大量极小文件是否存在最优实现方案？

小文件问题对S3写入性能的影响远大于读取，你当前的写法性能差核心是分区逻辑完全不合理，做了大量无用操作。

你当前代码repartition(col("id"))配合partitionBy(col("id"))的写法，在id为行级唯一值的场景下（也就是你要每行一个文件的需求），会直接生成和行数相等的写任务，10万行对应10万个任务，100万行对应100万个任务，开销来自三个层面：

• S3 API本身的开销：每个文件对应至少1次PUT请求、1次配套的权限校验、网络建连成本，S3单前缀默认PUT请求阈值是3500次/秒，10万次请求光排队就要近30秒，还不算请求失败重试的成本。而且你用的是overwrite模式，写之前Spark要遍历目标路径删除旧文件，小文件越多，列目录、删文件的元数据操作耗时越长。如果用的是原生S3A文件系统，每个文件写完还要做一次rename操作（本质是COPY+DELETE），小文件场景下这部分开销直接翻倍。
• Spark任务调度开销：每个写任务都要经过Driver调度、分发到Executor、序列化上下文、执行、资源回收的完整流程，10万个以上的任务会让Driver的状态维护压力陡增，大量时间耗在任务调度上，根本没把资源用在实际数据传输上，极端情况还会触发Driver OOM。
• 多余的shuffle开销：你在partitionBy之前额外做了repartition(col("id"))，但partitionBy本身就会按id值拆分输出路径，前置的repartition会触发一次全量数据shuffle，在id唯一的场景下，shuffle后每个分区只有1条数据，网络传输的开销完全是浪费。

按优先级从高到低排：

• 去掉冗余操作先拿到基础收益：直接删掉.repartition(col("id"))这行，partitionBy本身就会完成按id拆分输出的逻辑，不需要额外做重分区，这一步就能省掉一次全量shuffle的耗时。
• 放弃原生write接口，用mapPartitions批量写：如果必须严格保持每行一个独立JSON文件，不要用DataFrame原生的.write.json接口，把DataFrame的并行度设置为和集群Executor总核数的2_{3倍持平（比如集群总共有200核，就设为400}600个分区），然后用mapPartitions算子在每个分区内部遍历行数据，直接调用S3 SDK逐行写JSON文件。这种方式总任务数只有几百个，完全避免了调度上万个任务的开销，每个任务内可以复用S3连接，把API请求的建连开销平摊掉，写入速度能提升5~10倍。
  写的时候注意控制单Executor内的S3请求并发，不要打满S3的请求阈值，同时配置S3参数跳过冗余流程：

  spark.hadoop.fs.s3a.fast.upload=true
  spark.hadoop.fs.s3a.threads.max=20
  spark.hadoop.fs.s3a.connection.maximum=1000
  spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2
  

• 去掉不必要的分区目录：如果下游进程不依赖id=xxx/这种Hive风格的分区目录结构，直接把所有文件写到统一路径下，用id作为JSON文件名即可，省去Spark创建上万个分区目录的元数据开销。
• 和下游约定格式做批量合并（最优长期方案）：如果下游可以适配，不要强行做每行一个文件，把每1000~10000行合并成一个JSON文件，文件名带上对应id的范围，下游读取时再做拆分，这种方式写入性能可以提升两个数量级，后续读取的性能也会有质的提升，从根源上解决小文件问题。

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