Spark+Hudi写入文件系统性能极差且报错，求优化方案

记录被收集到Driver是否正常？ #

不正常。Hudi的Upsert操作虽会涉及索引查询，但正常情况下数据处理应在Executor端并行执行，不会将全量记录拉取到Driver序列化。你遇到的全量数据回Driver属于异常情况，直接触发了Driver结果大小超限的错误。

核心问题定位 #

1. 分区配置严重不合理：将唯一标识users_activity_id同时设为RecordKey和PartitionPath字段，600万条记录会生成600万个分区，触发巨量Task（140多万个），Driver需要处理海量分区元数据和Task结果，直接压垮Driver。
2. 写入操作选型错误：首次全量写入却使用Upsert操作，Upsert需要做索引查找和预合并，额外增加大量不必要的开销。
3. Spark Driver参数不足：默认spark.driver.maxResultSize为1GB，无法承载百万级Task的结果数据。

具体优化方案 #

（1）修正分区策略 #

放弃用唯一ID做分区，改用时间类字段（如按users_activity_create_date按天/小时分区）或其他低基数分组字段，大幅减少分区数量；若无需分区，直接移除分区配置：

hudi_options = {
        'hoodie.table.name': 'users_activity',
        'hoodie.datasource.write.recordkey.field': 'users_activity_id',
        # 示例：按日期字段dt分区
        'hoodie.datasource.write.partitionpath.field': 'dt',
        'hoodie.datasource.write.table.name': 'users_activity_result',
        # 首次全量写入用bulk_insert替代upsert
        'hoodie.datasource.write.operation': 'bulk_insert',
        'hoodie.datasource.write.precombine.field': 'users_activity_create_date',
    }

（2）匹配写入操作场景 #

• 首次全量写入：使用bulk_insert，跳过索引查找和预合并，直接批量写入，性能远高于Upsert。
• 后续增量更新：再切换为Upsert操作。

（3）调整Spark关键参数 #

• 增大Driver结果大小限制：

  spark.driver.maxResultSize 2g
  

• 控制Task数量，避免Task过多：

  # 设置每个Task处理128MB数据，减少Task总数
  spark.sql.files.maxPartitionBytes 134217728
  # 设置合理的并行度
  spark.default.parallelism 200
  

• 开启Hudi并行索引（针对Upsert场景）：

  hudi_options['hoodie.index.type'] = 'BLOOM'
  hudi_options['hoodie.index.bloom.filter.num_entries'] = '6000000'
  hudi_options['hoodie.index.bloom.filter.fpp'] = '0.0001'
  

  Bloom索引让索引查找在Executor端并行执行，避免Driver拉取全量数据。

（4）其他优化措施 #

• 合并小文件：读取Parquet后先合并分区，减少Task数量：

  df = spark.read.option('mergeSchema', 'true').parquet('s3://an/existing/directory/')
  # 重分区为合理数量，比如200
  df = df.repartition(200)
  

• 优化S3访问：确保集群与S3同区域，开启S3快速上传：

  spark.hadoop.fs.s3a.fast.upload true
  

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