使用pyspark.pandas写入Delta表时遇Checkpoint block未找到错误

问题详情 #

• 执行代码：combined_df_final.to_delta(file_path, mode='append')
• 报错信息：Checkpoint block rdd_1733_0 not found
• 触发场景：单独运行代码可成功，但在for循环中随机崩溃，处理特定DataFrame时必失败
• 集群环境：Azure Databricks 11.3 LTS（Apache Spark 3.3.0、Scala 2.12），Worker节点4-7个（内存256-448GB、核心128-224），Driver节点32GB内存、16核心，运行时版本11.3.x-scala2.12

解决方案 #

1. 替换本地检查点为分布式检查点 #

本地检查点依赖Executor本地存储，循环中Executor可能重启或存储被清理，导致块丢失。按以下步骤切换：

• 先设置分布式检查点目录（需为DBFS等分布式存储路径）：

  spark.sparkContext.setCheckpointDir("/dbfs/checkpoints/your_custom_dir")
  

• 对目标DataFrame执行检查点后再写入：

  # 将DataFrame转为RDD执行检查点，再转回DataFrame
  checkpointed_rdd = combined_df_final.rdd.checkpoint()
  combined_df_final = spark.createDataFrame(checkpointed_rdd)
  # 执行写入操作
  combined_df_final.to_delta(file_path, mode='append')
  

2. 优化循环内作业资源管理 #

循环中多次作业易引发资源竞争或状态混乱，需及时清理缓存：

• 每次写入完成后，释放当前DataFrame的缓存：

  combined_df_final.unpersist()
  

• 避免在循环内不必要的DataFrame缓存，仅在需要时缓存并及时释放。

3. 调整集群与作业参数 #

• 增大Executor内存开销：在集群配置中调高spark.executor.memoryOverhead，避免本地存储不足导致检查点块被清理。
• 拆分批次处理：若单次循环处理数据量过大，将数据拆分为更小批次，降低检查点压力。

4. 改用Spark原生API写入Delta表 #

PySpark Pandas的高层封装可能在循环场景下存在隐藏检查点逻辑，尝试用原生API替代：

# 将PySpark Pandas DataFrame转为Spark DataFrame
spark_df = combined_df_final.to_spark()
# 写入Delta表
spark_df.write.format("delta").mode("append").save(file_path)

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