如何在Databricks中将文件保存至Gen2存储的分区子文件夹

以下是两种适配你场景的实现方法，可根据数据特点选择：

方法一：基于数据分区字段自动写入（推荐，适用于数据有对应分区标识） #

如果你的CSV数据中包含一个字段（比如partition_id），其值正好匹配那8个子文件夹的名称，可利用Spark的partitionBy功能自动将数据写入对应子文件夹：

# 读取源CSV数据
df = spark.read.csv("/mnt/container/folder/your_source_data.csv", header=True, inferSchema=True)

# 按指定字段分区写入Parquet，自动匹配子文件夹
df.write.mode("overwrite").parquet("/mnt/container/folder/", partitionBy="partition_id")

说明：此方式会自动创建partition_id=xxx格式的子文件夹，如果你的现有子文件夹不是这种命名，需先调整子文件夹名称，或修改数据中对应字段的值匹配现有名称。

方法二：手动拆分数据并逐个写入固定子文件夹 #

如果数据没有对应分区字段，或者需要写入已存在的固定命名子文件夹，可手动拆分数据后分别写入：

步骤1：获取目标子文件夹列表 #

先确认挂载路径下的子文件夹是否可访问：

# 列出folder下的所有子文件夹
subfolder_paths = [f.path for f in dbutils.fs.ls("/mnt/container/folder/") if f.isDir()]

步骤2：拆分数据并写入 #

根据业务逻辑拆分DataFrame（示例为按行数平均拆分），再逐个写入对应子文件夹：

# 读取源CSV数据
df = spark.read.csv("/mnt/container/folder/your_source_data.csv", header=True, inferSchema=True)

# 计算每个子文件夹要写入的数据量
total_rows = df.count()
rows_per_folder = total_rows // 8
remaining_rows = total_rows % 8

# 拆分数据并写入
current_df = df
for idx, folder_path in enumerate(subfolder_paths):
    # 分配数据量，最后一个文件夹处理剩余行数
    if idx == len(subfolder_paths) - 1:
        write_df = current_df.limit(rows_per_folder + remaining_rows)
    else:
        write_df = current_df.limit(rows_per_folder)
    
    # 写入Parquet到目标子文件夹
    write_df.write.mode("overwrite").parquet(folder_path)
    
    # 更新剩余数据
    current_df = current_df.subtract(write_df)

关键注意事项 #

• 写入模式：mode("overwrite")会覆盖目标文件夹内的现有内容，若需追加数据可改为mode("append")，根据业务需求调整。
• 权限检查：确保Databricks集群拥有Gen2存储容器及子文件夹的写入权限，权限配置需正确关联SAS密钥或服务主体。
• 路径准确性：确认子文件夹路径拼写无误，避免因路径错误导致文件写入到folder层级。

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