PySpark高效写入30亿条Parquet记录方案求助

1. 调整Spark作业资源配置 #

心跳错误多源于资源不足导致的executor超时，需根据集群规模调整核心参数：

• 增加executor内存与核心数，避免OOM或任务积压：

  spark-submit \
    --executor-memory 16G \
    --executor-cores 4 \
    --driver-memory 8G \
    --num-executors 64 \
    your_script.py
  

• 启用动态资源分配，让集群自动适配任务需求：

  spark.conf.set("spark.dynamicAllocation.enabled", "true")
  spark.conf.set("spark.dynamicAllocation.minExecutors", "8")
  spark.conf.set("spark.dynamicAllocation.maxExecutors", "128")
  

2. 合理规划数据分区 #

30亿条记录需避免单分区数据量过大（导致IO超时）或分区过多（产生大量小文件）：

• 按目标文件大小计算分区数：Parquet单文件建议128MB-256MB，假设单条记录约100字节，30亿条合计300GB，需1200-2400个分区：

  dfBomtransform = dfBomtransform.repartition(1800)
  

• 若业务允许，按维度字段分区（如时间、类别），分散写入压力：

  dfBomtransform.write \
    .format("parquet") \
    .partitionBy("biz_date") \
    .save('/mnt/fulldatabases/GES/serv')
  

3. 优化Parquet写入参数 #

通过参数调优降低IO开销、提升写入效率：

• 启用Snappy压缩（平衡压缩比与性能）：

  dfBomtransform.write \
    .format("parquet") \
    .option("compression", "snappy") \
    .mode("overwrite") \
    .save('/mnt/fulldatabases/GES/serv')
  

• 开启向量化写入（默认已开启，强制确认）：

  spark.conf.set("spark.sql.parquet.enableVectorizedWriter", "true")
  

• 调整shuffle分区数（适配大数量级数据）：

  spark.conf.set("spark.sql.shuffle.partitions", "2000")
  

4. 优化嵌套JSON展开逻辑 #

自定义flattenAndExplode可能存在性能瓶颈，优先用Spark内置函数替代：

• 用explode/inline处理嵌套数组，用select *展开结构体：

  from pyspark.sql.functions import explode, col
  
  # 第一步：展开顶层数组
  df_exploded = dfBomservices.select(explode(col("nested_array")).alias("item"))
  # 第二步：展开结构体字段
  df_flattened = df_exploded.select("item.*")
  

• 避免不必要的UDF，内置函数基于Java实现，性能远高于Python UDF。

5. 缓解数据倾斜 #

若展开后存在单key数据量过大的情况，通过加盐打散数据：

from pyspark.sql.functions import rand, floor

# 生成0-99的随机盐，打散倾斜key
df_with_salt = dfBomtransform.withColumn("salt", floor(rand() * 100))
# 按盐分区，避免单分区压力过大
df_repartitioned = df_with_salt.repartition("salt")
# 写入时丢弃盐字段
df_repartitioned.drop("salt").write.format("parquet").save('/mnt/fulldatabases/GES/serv')

6. 验证存储层IO能力 #

确认目标存储（如ADLS Gen2）的吞吐量与带宽达标：

• 若使用云存储，检查存储账户的性能层级（如Premium层）是否满足大流量写入需求；
• 避免跨区域写入，减少网络延迟。

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