在Databricks上转换Parquet到Delta Lake遇FileAlreadyExistsException异常

问题描述 #

尝试在Databricks上将Azure Data Lake Storage（ADLS）中的Parquet文件原地转换为Delta格式（仅添加_delta_log，避免复制数据），转换前会检查并删除已存在的_delta_log目录，但即使目录不存在，执行CONVERT TO DELTA仍抛出FileAlreadyExistsException。

代码与错误信息 #

出错的转换代码 #

from pyspark.sql import SparkSession

def cleanup_delta_log(staging_path):
    delta_log_path = f"{staging_path}/_delta_log"
    spark = SparkSession.builder.getOrCreate()
    
    if spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration()).exists(spark._jvm.org.apache.hadoop.fs.Path(delta_log_path)):
        print(f"Deleting existing _delta_log at {delta_log_path}")
        spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration()).delete(spark._jvm.org.apache.hadoop.fs.Path(delta_log_path), True)

def process_table(table_name, source_name):
    staging_path = f"/mnt/landing/{source_name}/{table_name}/{table_name}"
    print(f"staging_path is {staging_path}")

    spark = SparkSession.builder.appName(f"Processing_Table_{table_name}").getOrCreate()

    cleanup_delta_log(staging_path)
    
    spark.sql(f"CONVERT TO DELTA parquet.`{staging_path}`")

# Process tables...

错误信息 #

staging_path is /mnt/landing/source_name/tablename/tablename
org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration()).exists(spark._jvm.org.apache.hadoop.fs.Path(delta_log_path)):
        print(f"Deleting existing _delta_log at {delta_log_path}")
        spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration()).delete(spark._jvm.org.apache.hadoop.fs.Path(delta_log_path), True)

def process_table(table_name, source_name):
    staging_path = f"/mnt/landing/{source_name}/{table_name}/{table_name}"
    print(f"staging_path is {staging_path}")

    spark = SparkSession.builder.appName(f"Processing_Table_{table_name}").getOrCreate()

    cleanup_delta_log(staging_path)
    
    spark.sql(f"CONVERT TO DELTA parquet.`{staging_path}`")

# Process tables...

可正常工作但会复制数据的替代方案 #

df = spark.read.format("parquet").load(staging_path)
df.write.format("delta").mode("overwrite").saveAsTable(
    f"{source_name}.{source_name}_{table_name}"
)

已验证事项 #

• 转换前目标路径下确实不存在_delta_log目录
• 拥有ADLS的读写权限

问题原因与解决方案 #

可能原因 #

1. 路径中存在非Parquet文件：CONVERT TO DELTA要求目标路径仅包含.parquet后缀文件，若存在临时文件、日志文件等无关文件，会触发错误。
2. Metastore残留表记录：目标路径曾被注册为Delta表，即使手动删除_delta_log，Metastore中仍有残留记录，导致转换冲突。
3. 文件系统元数据延迟：手动删除_delta_log后，ADLS的元数据未及时同步，Spark仍判定目录存在。

对应解决方案 #

方案1：清理非Parquet文件

检查并删除目标路径下的非Parquet文件：

def cleanup_non_parquet_files(staging_path):
    spark = SparkSession.builder.getOrCreate()
    fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())
    path = spark._jvm.org.apache.hadoop.fs.Path(staging_path)
    
    for file_status in fs.listStatus(path):
        if not file_status.getPath().getName().endswith(".parquet"):
            print(f"Deleting non-parquet file: {file_status.getPath()}")
            fs.delete(file_status.getPath(), False)

# 在cleanup_delta_log之后调用
cleanup_non_parquet_files(staging_path)

方案2：注销Metastore残留表

若路径曾注册为表，先删除Metastore中的记录：

spark.sql(f"DROP TABLE IF EXISTS {source_name}.{table_name}")

执行后再重新运行CONVERT TO DELTA。

方案3：使用Delta原生API清理

避免手动操作HDFS文件系统，改用Delta Lake API保证元数据一致性：

from delta.tables import DeltaTable

def cleanup_delta_log(staging_path):
    if DeltaTable.isDeltaTable(spark, staging_path):
        print(f"Converting existing path back to parquet first")
        delta_table = DeltaTable.forPath(spark, staging_path)
        delta_table.convertToParquet()

方案4：使用OVERWRITE参数（Delta Lake 2.0+支持）

若使用Delta Lake 2.0及以上版本，可在转换语句中添加OVERWRITE强制覆盖状态：

spark.sql(f"CONVERT TO DELTA parquet.`{staging_path}` OVERWRITE")

总结 #

优先检查并清理路径中的非Parquet文件，这是最常见的触发原因。若问题仍存在，再排查Metastore残留记录或使用Delta原生API清理。OVERWRITE参数是最简单的强制解决方式（需对应版本支持）。

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