使用Dataproc Serverless将PySpark DataFrame写入BigQuery失败求助

问题背景 #

• 在Dataproc无服务器批处理作业中，完成从BigQuery/Cloud Storage读取数据、特征工程（含开销较大的Join和CrossJoin操作）后，尝试将约33M行的PySpark DataFrame写入BigQuery时失败。

报错信息 #

22/10/08 08:13:21 WARN BigQueryDataSourceWriterInsertableRelation: It seems that 184 out of 16 partitions have failed, aborting
22/10/08 08:13:21 WARN BigQueryDirectDataSourceWriterContext: BigQuery Data Source writer aedb4dc8-28c5-4118-9dcc-de2ef689e75c aborted

当前Spark配置 #

--properties spark.executor.instances=10,spark.driver.cores=16,spark.executor.cores=16

写入代码 #

user_item_interaction_df.write.format("bigquery").option("writeMethod", "direct").mode("overwrite").save()

解决思路 #

1. 调整Spark分区与资源配置 #

• 报错中“184 out of 16 partitions”的异常说明分区数与任务负载不匹配，大概率是CrossJoin导致数据倾斜，部分分区数据量过大。先检查当前DataFrame分区数：

  print(user_item_interaction_df.rdd.getNumPartitions())
  

• 如果分区数过少，重分区拆分大分区：

  # 根据数据量调整，建议按每分区100-500K行设置，33M数据可设为100-200分区
  user_item_interaction_df = user_item_interaction_df.repartition(150)
  

• 优化Spark资源配置，补充内存参数并调整核数配比（Dataproc无服务器核内存配比建议1核对应2G）：

  --properties spark.executor.instances=10,spark.driver.cores=8,spark.driver.memory=16g,spark.executor.cores=8,spark.executor.memory=16g,spark.sql.shuffle.partitions=200
  

2. 优化BigQuery写入策略 #

• 切换为indirect写入方式（先写GCS再导入BigQuery），稳定性远高于direct，适合大数据量：

  user_item_interaction_df.write.format("bigquery")
    .option("writeMethod", "indirect")
    .option("temporaryGcsBucket", "your-temp-gcs-bucket") # 替换为你的GCS临时桶
    .mode("overwrite")
    .save()
  

• 若坚持用direct，增加重试机制降低写入失败概率：

  user_item_interaction_df.write.format("bigquery")
    .option("writeMethod", "direct")
    .option("bigqueryRetryTimes", "5")
    .option("bigqueryRetryDelay", "10") # 重试间隔秒数
    .mode("overwrite")
    .save()
  

3. 解决CrossJoin导致的数据倾斜 #

• 先过滤无效数据再执行CrossJoin，减少总数据量：比如过滤掉无交互记录的用户/物品。
• 对关联的小表使用广播变量，避免全量shuffle：

  from pyspark.sql.functions import broadcast
  # 假设small_df是较小的表，large_df是大表
  user_item_interaction_df = broadcast(small_df).crossJoin(large_df)
  

• 抽样检查数据分布，若存在极端倾斜的key，拆分这些key单独处理后再合并结果。

4. 定位具体失败原因 #

• 登录Dataproc控制台打开作业的Spark UI，查看失败任务的stderr日志，确认是OOM、网络超时还是BigQuery API限流。
• 检查GCP控制台的BigQuery配额，确认是否因写入请求频次过高触发限流，必要时申请配额提升。

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