Azure Databricks挂载点可访问但保存文件提示目录不存在

问题原因 #

你通过dbutils.fs.mount创建的挂载点属于Databricks文件系统（DBFS）路径，但Pandas的to_csv方法直接操作的是集群节点的本地文件系统——本地节点上不存在/mnt/MyMount这个目录，因此触发报错。而dbutils.fs.ls能正常访问，是因为它是DBFS专属操作接口，和本地文件系统不属于同一空间。

解决方案 #

方案1：使用Spark API直接保存CSV（推荐） #

避免转换为Pandas DataFrame，直接用Spark写入API操作DBFS挂载点，性能更优且适配Databricks环境：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("MyApp").getOrCreate()
views = ['view_1', 'view_2', 'view_3', 'view_4', 'view_5']
logs = ['Log_views']

for view in views:
    print(f'Start save view {view}.')
    spark.read.format("delta").load(f"/mnt/source/{view}") \
        .write.format("csv") \
        .option("header", "true") \
        .mode("overwrite") \
        .save(f"/mnt/MyMount/{view}")
    print(f'View {view} save successfully.')

for log in logs:
    print(f'Start save log {log}.')
    spark.read.format("parquet").load(f"/mnt/source/{log}") \
        .write.format("csv") \
        .option("header", "true") \
        .mode("overwrite") \
        .save(f"/mnt/MyMount/{log}")
    print(f'Logs {log} save successfully.')

如果需要生成单个CSV文件，可先合并分区再处理：

# 以单个view为例
view = "view_1"
spark.read.format("delta").load(f"/mnt/source/{view}") \
    .coalesce(1)  # 合并为1个分区，确保生成单个文件
    .write.format("csv") \
    .option("header", "true") \
    .mode("overwrite") \
    .save(f"/mnt/MyMount/{view}_temp")

# 重命名临时文件并清理临时目录
temp_files = dbutils.fs.ls(f"/mnt/MyMount/{view}_temp")
csv_file = next(file.path for file in temp_files if file.name.endswith(".csv"))
dbutils.fs.mv(csv_file, f"/mnt/MyMount/{view}.csv")
dbutils.fs.rm(f"/mnt/MyMount/{view}_temp", recurse=True)

方案2：通过DBFS FUSE挂载路径使用Pandas写入 #

Databricks将DBFS通过FUSE挂载到集群节点的/dbfs目录下，你可以通过这个前缀让Pandas访问挂载点：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("MyApp").getOrCreate()
views = ['view_1', 'view_2', 'view_3', 'view_4', 'view_5']
logs = ['Log_views']

for view in views:
    print(f'Start save view {view}.')
    spark.read.format("delta").load(f"/mnt/source/{view}").toPandas() \
        .to_csv(f"/dbfs/mnt/MyMount/{view}.csv", index=False)
    print(f'View {view} save successfully.')

for log in logs:
    spark.read.format("parquet").load(f"/mnt/source/{log}").toPandas() \
        .to_csv(f"/dbfs/mnt/MyMount/{log}.csv", index=False)
    print(f'Logs {log} save successfully.')

注意：该方式仅适合小数据集，大数据集转换为Pandas会占用大量节点内存，易引发内存溢出。

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