如何使用Databricks实现Azure Files与ADLS Gen 2双向文件迁移

整体分3步完成：先将两个存储服务挂载到Databricks文件系统（DBFS），再实现双向迁移逻辑，最后配置定时调度即可。

1. 前置配置：挂载存储到DBFS #

先将Azure Files和ADLS Gen2分别挂载到Databricks的统一访问路径，后续可以直接用通用文件操作命令读写，不需要每次传入鉴权信息。存储密钥建议存在Databricks Secrets中，不要硬编码在代码里。

• 挂载Azure文件共享示例代码

storage_account_name = "你的存储账号名"
file_share_name = "你的文件共享名"
storage_account_key = dbutils.secrets.get(scope="你的密钥作用域", key="存储账号密钥的key名")
mount_point = "/mnt/azure_files/text_output"

# 避免重复挂载报错
if not any(mount.mountPoint == mount_point for mount in dbutils.fs.mounts()):
    dbutils.fs.mount(
      source = f"wasbs://{file_share_name}@{storage_account_name}.file.core.windows.net/",
      mount_point = mount_point,
      extra_configs = {f"fs.azure.account.key.{storage_account_name}.file.core.windows.net": storage_account_key}
    )

• 挂载ADLS Gen2容器示例代码

storage_account_name = "你的ADLS存储账号名"
container_name = "你的ADLS容器名"
storage_account_key = dbutils.secrets.get(scope="你的密钥作用域", key="ADLS密钥的key名")
mount_point = "/mnt/adls_gen2/text_archive"

if not any(mount.mountPoint == mount_point for mount in dbutils.fs.mounts()):
    dbutils.fs.mount(
      source = f"abfss://{container_name}@{storage_account_name}.dfs.core.windows.net/",
      mount_point = mount_point,
      extra_configs = {f"fs.azure.account.key.{storage_account_name}.dfs.core.windows.net": storage_account_key}
    )

2. 双向迁移逻辑实现 #

2.1 Azure Files 迁移到 ADLS Gen2 #

支持全量和增量两种模式，增量模式可以避免重复拷贝文件，适合定时执行的场景：

import time
# 增量迁移参数：只迁移最近24小时生成的文件，时间可自定义
cutoff_timestamp = time.time() * 1000 - 24 * 3600 * 1000 

source_path = "/mnt/azure_files/text_output/"
target_path = "/mnt/adls_gen2/text_archive/"

for file in dbutils.fs.ls(source_path):
    # 过滤符合时间要求的文件
    if file.modificationTime > cutoff_timestamp and file.name.endswith(".txt"):
        # 拷贝文件，recurse=True支持目录递归拷贝
        dbutils.fs.cp(file.path, target_path + file.name, recurse=True)
        # 可选操作：迁移完成后删除源文件，或者移动到已归档目录打标记
        # dbutils.fs.rm(file.path, recurse=True)

如果是超大量文件迁移，建议用Spark分布式读写，性能远高于单节点的dbutils.fs.cp：

# 分布式全量迁移示例
df = spark.read.text(source_path + "*.txt")
df.write.mode("append").text(target_path)

2.2 ADLS Gen2 回迁到 Azure Files #

逻辑和正向迁移完全一致，只需要交换源路径和目标路径即可，支持按文件名、时间范围筛选需要回迁的文件：

# 按日期回迁示例：回迁2024年6月的所有归档文件
restore_source = "/mnt/adls_gen2/text_archive/202406/*.txt"
restore_target = "/mnt/azure_files/restore/202406/"

df = spark.read.text(restore_source)
df.write.mode("overwrite").text(restore_target)

3. 固定周期调度配置 #

直接用Databricks自带的Workflow调度功能即可：

• 将迁移代码整理为独立的Notebook或者Python脚本
• 在Databricks工作台左侧进入「Workflows」→「Jobs」→ 新建Job
• 选择对应的计算集群，设置触发周期（比如每天凌晨1点执行），可配置失败重试次数、运行告警规则

可选优化点 #

• 迁移后增加校验逻辑：对比源端和目标端的文件数量、总大小，避免文件丢失
• 大量小文件场景可以先合并为大文件再迁移，降低存储成本和后续查询耗时
• 超大文件可开启压缩后再传输，减少迁移耗时

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