如何通过dbutils.fs.mv快速迁移Azure存储中的百万级文件？

针对单个文件夹下超百万文件的迁移需求，基于dbutils.fs.mv工具的最快实现核心是最大化并行处理能力，同时规避单节点/单线程瓶颈，以下是具体方案：

1. 配置高性能Databricks集群 #

• 采用多节点计算优化型集群（如DSv3/Fsv2系列实例），节点数量根据文件规模调整（建议8-16节点起步，每节点4-8核），利用Executor的分布式算力并行处理文件迁移。
• 确保集群与Azure存储账户处于同一区域，避免跨区域传输的网络延迟和额外成本。

2. 分布式获取文件路径，规避Driver内存溢出 #

不要用dbutils.fs.ls一次性拉取百万文件路径（会导致Driver内存爆掉），改用Spark分布式API批量读取文件路径：

from pyspark.sql.functions import input_file_name

# 仅读取文件路径，不加载文件内容，避免内存占用
file_paths_df = spark.read.text(
    "abfss://<container>@<storage-account>.dfs.core.windows.net/source-folder/*",
    wholetext=True
).select(input_file_name().alias("path"))

3. 分区并行执行迁移操作 #

将文件路径按Executor分区拆分，在每个分区内批量执行dbutils.fs.mv，让多个Executor同时工作：

def batch_move_files(file_path_batch):
    for src_path in file_path_batch:
        # 替换路径中的源文件夹为目标文件夹
        dest_path = src_path.replace("source-folder", "dest-folder")
        # 执行迁移，关闭递归（因为是单个文件）
        dbutils.fs.mv(src_path, dest_path, recurse=False)

# 将路径转为RDD，按分区批量处理
file_paths_df.rdd.map(lambda row: row.path).foreachPartition(batch_move_files)

• 这种方式让每个Executor独立处理一批文件，Driver仅负责分发任务，不会出现内存瓶颈。

4. 利用Azure存储优化提升IO性能 #

• 如果使用ADLS Gen2存储，确保已开启分层命名空间，提升小文件操作的效率。
• 临时切换存储账户到高级性能层级（Premium Blob Storage），迁移完成后再切回标准层级，小文件密集场景下IOPS会大幅提升。

5. 批量迁移优化（针对有规律的文件名） #

如果文件名有统一前缀/后缀，直接用通配符批量迁移，减少单文件mv的开销：

# 按前缀批量迁移，比如所有以"log-2024-"开头的文件
dbutils.fs.mv(
    "abfss://<container>@<storage-account>.dfs.core.windows.net/source-folder/log-2024-*",
    "abfss://<container>@<storage-account>.dfs.core.windows.net/dest-folder/",
    recurse=False
)

• 这种方式比单文件迁移效率高数倍，适合文件名有规律的场景。

6. 迁移后校验 #

用Spark分布式统计源和目标文件夹的文件数、总大小，确保迁移完整：

from pyspark.sql.functions import count, sum, length, col

# 统计源文件夹
source_stats = spark.read.text(
    "abfss://<container>@<storage-account>.dfs.core.windows.net/source-folder/*",
    wholetext=True
).agg(
    count(input_file_name()).alias("source_file_count"),
    sum(length(col("value"))).alias("source_total_size")
).collect()[0]

# 统计目标文件夹
dest_stats = spark.read.text(
    "abfss://<container>@<storage-account>.dfs.core.windows.net/dest-folder/*",
    wholetext=True
).agg(
    count(input_file_name()).alias("dest_file_count"),
    sum(length(col("value"))).alias("dest_total_size")
).collect()[0]

print(f"源文件数：{source_stats.source_file_count} | 目标文件数：{dest_stats.dest_file_count}")
print(f"源总大小：{source_stats.source_total_size} | 目标总大小：{dest_stats.dest_total_size}")

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