PySpark checkpointed DF复用及相关运维问题咨询

问题1：checkpoint完成后作业失败，重新提交如何恢复？ #

默认的checkpoint()属于非可靠检查点，不会自动复用已生成的检查点数据，要实现失败恢复，需手动管理检查点的加载逻辑：

• 提前规划好检查点的存储路径（避免默认的随机目录），作业重启时先检查该路径是否存在有效数据：

  # 示例恢复逻辑
  target_checkpoint_path = "hdfs://xxx/your-fixed-checkpoint-path"
  try:
      # 尝试从检查点加载DataFrame
      df5 = spark.read.parquet(target_checkpoint_path)
  except Exception:
      # 检查点不存在，重新执行转换流程
      df4 = ... # 你的原始转换逻辑
      # 将df4写入指定路径作为自定义检查点
      df4.write.mode("overwrite").parquet(target_checkpoint_path)
      df5 = spark.read.parquet(target_checkpoint_path)
  

• 若坚持使用原生checkpoint()，则需要在生成检查点时记录下实际的存储路径（可通过df5.rdd.getCheckpointFile()获取），重启时直接读取该路径下的parquet文件即可。

问题2：PySpark作业完成后如何清理checkpoint文件？ #

有三种可行方式，无需单独依赖外部bash脚本：

1. 作业内用HDFS API清理：针对自定义的检查点路径，直接调用Hadoop文件系统API删除：

   from pyspark.sql import SparkSession
   hadoop_conf = spark.sparkContext._jsc.hadoopConfiguration()
   fs = spark.sparkContext._jvm.org.apache.hadoop.fs.FileSystem.get(hadoop_conf)
   checkpoint_path = "hdfs://xxx/your-checkpoint-path"
   target_path = spark.sparkContext._jvm.org.apache.hadoop.fs.Path(checkpoint_path)
   if fs.exists(target_path):
       fs.delete(target_path, True) # True表示递归删除子目录
   

2. 使用Spark内置方法清理：若用默认setCheckpointDir生成的随机目录，可调用spark.sparkContext.clearCheckpointDir()清理整个根目录下的所有检查点文件，但需注意会删除该目录下所有内容。
3. Airflow编排时添加清理步骤：在DAG中新增BashOperator，执行HDFS删除命令：

   hdfs dfs -rm -r hdfs://xxx/your-checkpoint-dir/*
   

   适合在作业成功完成后触发清理。

问题3：能否自定义checkpoint目录下的文件夹名称？ #

原生checkpoint()方法会自动生成{checkpoint dir}/{随机数字}/rdd-{随机数字}格式的随机路径，无法直接自定义这些自动生成的文件夹名称。
替代方案是放弃原生checkpoint()，改为手动将DataFrame写入自定义命名的路径，完全自主控制存储目录的名称，同时也更便于后续的恢复和清理操作，示例可参考问题1中的手动写入逻辑。

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