DSEFS删除parquet文件报锁占用/目录非空错误求解

问题根因 #

报错核心原因是之前运行的Python/PySpark parquet写入任务异常退出，没有释放DSEFS在_temporary临时写入目录上的分布式锁。手动执行rm -r命令时无法获取对应锁权限，下层临时目录删除失败，才会逐层抛出目录非空错误。

手动清理残留的操作步骤 #

• 排查并清理残留进程
  执行dse fs job list查找报错信息中标记的锁持有进程ID（即报错文本里possibly owned by Set(xxx)括号内的UUID），如果对应任务处于僵死、失败状态，直接kill对应进程，等待30秒让DSEFS自动回收锁资源。
• 强制清理临时目录残留
  不要直接删除parquet根目录，优先递归强制删除临时写入目录：

  dse fs -rm -r -f /work/test.parquet/_temporary
  

  如果上述命令仍然报锁错误，先逐层列出子目录下的锁文件，优先删除.lock后缀的锁文件后再执行临时目录删除：

  dse fs -ls /work/test.parquet/_temporary/0/_temporary/
  # 对列出的每个.lock文件执行强制删除
  dse fs -rm -f <lock_file_absolute_path>
  

• 删除目标parquet目录
  临时目录清理完成后，执行dse fs -rm -r /work/test.parquet即可正常删除整个parquet路径，不会再触发目录非空报错。

免手动操作的自动覆盖写入方案 #

不需要每次手动删文件或者重命名路径，在PySpark写入代码中做如下配置即可从根源规避问题：

• 提交Spark任务时增加锁自动回收和提交器优化配置，僵死任务持有的锁30秒自动回收，同时使用v2版本的文件提交器自动清理写入失败的残留文件：

  from pyspark.sql import SparkSession
  
  spark = SparkSession.builder \
      .config("spark.hadoop.dsefs.locking.lease.recovery.timeout", "30000") \
      .config("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2") \
      .getOrCreate()
  

• 写入前优先使用Hadoop FS API删除目标路径，该接口原生适配DSEFS的锁逻辑，比shell的rm命令兼容性更好：

  hadoop_conf = spark._jsc.hadoopConfiguration()
  fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(hadoop_conf)
  target_path = spark._jvm.org.apache.hadoop.fs.Path("/work/test.parquet")
  if fs.exists(target_path):
      fs.delete(target_path, True) # 参数True代表递归删除路径下所有内容
  
  # 后续正常执行parquet写入逻辑即可
  # df.write.parquet("/work/test.parquet")
  

• 高可靠场景可以采用临时路径写入+原子替换的模式：先将数据写入和目标路径无关的临时目录，写入完成校验通过后，用FS API的rename操作原子替换目标路径，就算写入中途失败，也不会在目标路径残留锁和临时文件。

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