如何使用Azure Data Factory实现YYYY-MM格式月度文件夹的自动拷贝

针对你Data Lake中每月生成的YYYY-MM格式文件夹（如2022-11），需要自动将文件夹内文件拷贝至其他路径的需求，以下是三种可行的实现方案：

方案一：Azure Logic Apps（低代码快速实现） #

适合无需编写代码的场景，通过可视化配置完成：

• 设置触发方式：
  • 定时触发：用Recurrence触发器，配置为每月固定日期（如每月1号凌晨）执行，覆盖当月生成的文件夹。
  • 事件触发：用Blob Storage触发器，监听源容器的文件夹创建事件，当新的YYYY-MM格式文件夹生成时立即触发。
• 筛选有效文件夹：添加Condition操作，用正则表达式^\d{4}-(0[1-9]|1[0-2])$判断文件夹名称是否符合标准月度格式，过滤无效文件夹。
• 遍历与拷贝文件：
  1. 用List Blobs操作列出目标文件夹下的所有文件。
  2. 用For Each循环遍历每个文件，执行Copy Blob操作，指定目标Data Lake的目标路径（如目标容器/归档/{文件夹名}）。
• 容错处理：添加Retry Policy设置失败重试次数，或配置邮件通知，确保拷贝异常可及时发现。

方案二：Azure Data Factory（ETL场景适配） #

适合已有数据工厂体系，需要整合进现有数据流水线的场景：

• 构建核心管道：
  1. 添加Get Metadata活动，获取源容器下的所有文件夹列表。
  2. 用Filter活动，通过正则表达式筛选出符合YYYY-MM格式的文件夹。
  3. 用For Each活动循环处理每个筛选后的文件夹。
  4. 在循环内添加Copy Data活动，源数据集指向当前遍历的文件夹（开启递归读取），目标数据集指向指定的目标路径。
• 配置触发机制：
  • 定时触发：创建Schedule Trigger，设置每月执行一次。
  • 事件触发：配置Blob Event Trigger，监听源容器的文件夹创建事件，实时触发拷贝。
• 增量控制：可在Data Lake中维护一个已处理文件夹清单文件（如processed_folders.txt），每次执行前读取清单，跳过已处理的文件夹，避免重复拷贝。

方案三：Azure Functions（自定义逻辑实现） #

适合需要高度自定义拷贝逻辑的场景：

• 选择触发器：
  • Timer Trigger：设置为每月执行一次，批量处理当月生成的文件夹。
  • Blob Trigger：监听源容器的文件夹创建事件，实时处理新生成的月度文件夹。
• 编写核心代码：
  以Python为例，使用azure-storage-blobs SDK：

  from azure.storage.blob import BlobServiceClient
  import re
  
  def copy_monthly_files(source_container, target_container, connection_string):
      blob_service_client = BlobServiceClient.from_connection_string(connection_string)
      source_container_client = blob_service_client.get_container_client(source_container)
      
      # 筛选YYYY-MM格式的文件夹
      folder_pattern = re.compile(r'^\d{4}-(0[1-9]|1[0-2])$')
      folders = [blob.name.split('/')[0] for blob in source_container_client.list_blobs() if '/' in blob.name]
      valid_folders = list(set([f for f in folders if folder_pattern.match(f)]))
      
      for folder in valid_folders:
          # 遍历文件夹内所有文件
          blobs = source_container_client.list_blobs(name_starts_with=f"{folder}/")
          for blob in blobs:
              # 拷贝文件到目标路径
              target_blob_client = blob_service_client.get_blob_client(target_container, blob.name)
              source_blob_client = source_container_client.get_blob_client(blob)
              target_blob_client.start_copy_from_url(source_blob_client.url)
  

• 权限配置：给Function App分配Storage Blob Data Contributor角色，确保拥有源容器的读取权限和目标容器的写入权限。

通用注意事项 #

• 权限验证：确保执行任务的身份（托管身份/服务主体）拥有对应Data Lake容器的读写权限，避免因权限不足导致拷贝失败。
• 格式校验：使用严谨的正则表达式^\d{4}-(0[1-9]|1[0-2])$，避免处理2022-13这类无效格式的文件夹。
• 增量处理：通过维护已处理清单或记录处理时间戳，避免重复拷贝已处理过的文件和文件夹。

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