路径已存在却触发AnalysisException？Spark代码逻辑疑问

执行代码时触发错误：AnalysisException: path dbfs:/mnt/Abinandhana/Mart/Dimension/dimCustomer_parquet already exists

代码中通过os.path.exists判断路径，若路径存在则进入else块执行数据合并后覆盖写入逻辑，但仍出现上述报错。代码如下：

if not os.path.exists('/mnt/Abinandhana/Mart/Dimension/dimCustomer_parquet'):
    Customer_Mart.write.parquet('/mnt/Abinandhana/Mart/Dimension/dimCustomer_parquet')
else:
    df2 = Customer_Mart
    df1 = spark.read.parquet('/mnt/Abinandhana/Mart/Dimension/dimCustomer_parquet')

    merged_df = df2.alias("t2").join(df1.alias("t1"),(col("`t1`.`SourceSystemId`") == col("`t2`.`SourceSystemId`")), "left") \
    .select(
        col("`t2`.`CustomerId`"),
        col("`t2`.`SourceSystemId`"),
        coalesce("t2.AccountNo", "t1.AccountNo").alias("AccountNo"),
        coalesce("t2.AccountStatus", "t1.AccountStatus").alias("AccountStatus"),
        coalesce("t2.AccountDesc", "t1.AccountDesc").alias("AccountDesc"),
        coalesce("t2.PrimaryGroupNo", "t1.PrimaryGroupNo").alias("PrimaryGroupNo"),
        coalesce("t2.PrimaryGroupDesc", "t1.PrimaryGroupDesc").alias("PrimaryGroupDesc"),
        coalesce("t2.Customer", "t1.Customer").alias("Customer"),
        coalesce("t2.CustomerGroup", "t1.CustomerGroup").alias("CustomerGroup"),
        coalesce("t2.DateOpened", "t1.DateOpened").alias("DateOpened"),
        coalesce("t2.DateClosed", "t1.DateClosed").alias("DateClosed")
    )
    display(merged_df)

    merged_df.write.mode("overwrite")\
    .parquet('/mnt/Abinandhana/Mart/Dimension/dimCustomer_parquet')

• os.path.exists不支持DBFS路径：os.path.exists是Python本地文件系统工具，无法识别Databricks的DBFS分布式文件系统路径。代码中判断逻辑实际未生效，当路径真实存在时，可能错误进入if分支执行无覆盖模式的写入，从而触发路径已存在的错误。
• 潜在竞态条件：即使os.path.exists能识别DBFS路径，在多任务并发场景下，判断路径不存在到执行写入的间隙，可能有其他进程创建该路径，导致写入冲突，但核心问题仍是前者。

方案1：用Spark/Hadoop工具判断DBFS路径 #

替换os.path.exists为支持DBFS的路径判断方法，比如通过Spark的Hadoop文件系统API：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, coalesce

def dbfs_path_exists(path):
    spark = SparkSession.getActiveSession()
    fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())
    return fs.exists(spark._jvm.org.apache.hadoop.fs.Path(path))

if not dbfs_path_exists('/mnt/Abinandhana/Mart/Dimension/dimCustomer_parquet'):
    Customer_Mart.write.parquet('/mnt/Abinandhana/Mart/Dimension/dimCustomer_parquet')
else:
    df2 = Customer_Mart
    df1 = spark.read.parquet('/mnt/Abinandhana/Mart/Dimension/dimCustomer_parquet')
    
    merged_df = df2.alias("t2").join(df1.alias("t1"),(col("`t1`.`SourceSystemId`") == col("`t2`.`SourceSystemId`")), "left") \
    .select(
        col("`t2`.`CustomerId`"),
        col("`t2`.`SourceSystemId`"),
        coalesce("t2.AccountNo", "t1.AccountNo").alias("AccountNo"),
        coalesce("t2.AccountStatus", "t1.AccountStatus").alias("AccountStatus"),
        coalesce("t2.AccountDesc", "t1.AccountDesc").alias("AccountDesc"),
        coalesce("t2.PrimaryGroupNo", "t1.PrimaryGroupNo").alias("PrimaryGroupNo"),
        coalesce("t2.PrimaryGroupDesc", "t1.PrimaryGroupDesc").alias("PrimaryGroupDesc"),
        coalesce("t2.Customer", "t1.Customer").alias("Customer"),
        coalesce("t2.CustomerGroup", "t1.CustomerGroup").alias("CustomerGroup"),
        coalesce("t2.DateOpened", "t1.DateOpened").alias("DateOpened"),
        coalesce("t2.DateClosed", "t1.DateClosed").alias("DateClosed")
    )
    display(merged_df)
    
    merged_df.write.mode("overwrite").parquet('/mnt/Abinandhana/Mart/Dimension/dimCustomer_parquet')

也可以用Databricks自带的dbutils工具：

# 检查目标路径是否存在
target_path = '/mnt/Abinandhana/Mart/Dimension/dimCustomer_parquet'
path_exists = any(f.path.rstrip('/') == target_path.rstrip('/') for f in dbutils.fs.ls('/mnt/Abinandhana/Mart/Dimension/'))

if not path_exists:
    Customer_Mart.write.parquet(target_path)
else:
    # 原else块逻辑不变

方案2：简化逻辑，直接用异常处理+覆盖模式 #

去掉路径判断，通过异常处理处理路径不存在的情况，直接使用overwrite模式写入：

from pyspark.sql.functions import col, coalesce
from pyspark.sql.utils import AnalysisException

target_path = '/mnt/Abinandhana/Mart/Dimension/dimCustomer_parquet'

try:
    # 尝试读取现有数据
    df1 = spark.read.parquet(target_path)
    df2 = Customer_Mart
    
    merged_df = df2.alias("t2").join(df1.alias("t1"),(col("`t1`.`SourceSystemId`") == col("`t2`.`SourceSystemId`")), "left") \
    .select(
        col("`t2`.`CustomerId`"),
        col("`t2`.`SourceSystemId`"),
        coalesce("t2.AccountNo", "t1.AccountNo").alias("AccountNo"),
        coalesce("t2.AccountStatus", "t1.AccountStatus").alias("AccountStatus"),
        coalesce("t2.AccountDesc", "t1.AccountDesc").alias("AccountDesc"),
        coalesce("t2.PrimaryGroupNo", "t1.PrimaryGroupNo").alias("PrimaryGroupNo"),
        coalesce("t2.PrimaryGroupDesc", "t1.PrimaryGroupDesc").alias("PrimaryGroupDesc"),
        coalesce("t2.Customer", "t1.Customer").alias("Customer"),
        coalesce("t2.CustomerGroup", "t1.CustomerGroup").alias("CustomerGroup"),
        coalesce("t2.DateOpened", "t1.DateOpened").alias("DateOpened"),
        coalesce("t2.DateClosed", "t1.DateClosed").alias("DateClosed")
    )
    display(merged_df)
    
    merged_df.write.mode("overwrite").parquet(target_path)
except AnalysisException:
    # 路径不存在时直接写入初始数据
    Customer_Mart.write.parquet(target_path)

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