Dataproc Serverless PySpark写入GCS复制阶段过慢问题求助

使用基于PySpark的Dataproc Serverless应用，作业逻辑如下：读取GCS上按d_date_sale分区的800+个5-10MB Parquet文件，转换后新增store_id分区维度写入GCS，生成的文件多为1-2MB。观察到写入临时目录.staging-spark-*速度较快，但后续复制步骤耗时是前者的3-4倍，需排查原因。

配置详情 #

version: 2.1
spark.dataproc.scaling.version: 2
spark.executor.cores: 8

代码片段 #

import functools
from pyspark.sql import DataFrame
import pyspark.sql.functions as F
import types

class CloudStorage(object):
    @staticmethod
    def _add_duplicate_partition_key(
        src_df: DataFrame,
        partition_key: str,
    ) -> DataFrame:
        duplicated_partition_key = f"_{partition_key}"

        return src_df.withColumn(
            colName=duplicated_partition_key, col=F.col(partition_key)
        )

    @staticmethod
    def write(
        df: DataFrame,
        partitions_cols: list[str] | None,
        dest_bucket: str,
        data_type: types.SupportedSourceTypes,
        mode: types.ParquetWriteModeTypes,
    ):
        # Spark removes the partitions cols, we want them to keep them
        # so we duplicate them and add prefix with _{colname}
        dup_part_key_df: DataFrame = functools.reduce(
            lambda res_df, p_key: CloudStorage._add_duplicate_partition_key(
                res_df, p_key
            ),
            partitions_cols,
            df,
        )

        # repartitioned = duplicated_partition_key_df.repartition(*[F.col(f"_{column}") for column in partitions_cols])

        dup_part_key_df.write.partitionBy(partitions_cols).mode(mode).option("compression", "snappy").format(
            'parquet').save(f"gs://{dest_bucket}/data/{data_type}/")

### Example call ### 
paths = [f"gs://{self.config.dest_bucket}/{prefix}*.parquet" for prefix in partition_dates]
df = self.spark.read.load(path=paths, format="parquet")
CloudStorage.write(df, cols, dest_bucket, ..)

1. 大量小文件导致复制阶段开销剧增 #

Spark写入分区表时，会先将数据写入临时目录，再通过原子性重命名/复制将文件移动到最终路径。你生成的文件仅1-2MB，远低于GCS最优文件大小（建议128MB-256MB），带来两个核心问题：

• 每个小文件都需要单独发起GCS API请求（复制、元数据更新），固定延迟累加后，整体耗时呈指数级增长；
• 小文件的IO吞吐量无法饱和executor的网络资源，大量时间消耗在连接建立、元数据查询等非数据传输环节。

2. 分区策略引发的文件数量爆炸 #

新增store_id分区后，若store_id基数较高，会导致总分区数（d_date_sale × store_id）大幅增加，每个分区下又有多个小文件，进一步放大复制阶段的处理压力。同时代码中注释掉了repartition逻辑，可能存在：

• 数据分布不均：部分executor生成的文件数量远多于其他节点，成为复制阶段的瓶颈；
• Task级别的小文件泛滥：每个写入Task默认生成多个小文件，没有通过预分区控制文件数量。

3. Dataproc Serverless与Spark版本的特性限制 #

• Spark 2.1版本局限性：较老的Spark版本对GCS文件系统的优化不足，缺乏批量复制/重命名等特性，处理大量小文件时效率低下；
• 配置资源不匹配：仅设置spark.executor.cores:8，未配置executor内存、实例数等参数，可能导致IO资源不足，无法支撑大量小文件的复制操作；
• 跨区域存储影响：若临时目录与最终目录不在同一GCS区域，跨区域复制会引入额外的带宽延迟和传输开销。

4. 写入模式的原子性开销 #

如果使用overwrite模式写入，Spark会先删除目标分区的原有文件，再执行临时文件复制，额外增加了删除大量小文件的开销；即使是append模式，大量小文件的元数据校验、目录更新操作也会消耗更多时间。

• 启用注释中的repartition逻辑，按store_id和d_date_sale预分区，结合数据量调整分区数，确保每个输出文件大小接近128MB；
• 设置spark.sql.files.maxRecordsPerFile参数，限制单个文件的记录数，从源头减少小文件生成；
• 确认临时目录与最终目录在同一GCS区域，避免跨区域传输；
• 升级Spark版本至3.x，新版本对GCS的文件操作有批量处理优化，能显著提升小文件场景下的复制效率；
• 调整Dataproc Serverless资源配置，增加executor内存或实例数，提升IO处理能力。

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